package k8score

import (
	"context"
	"fmt"
	"log"
	"sort"
	"strings"

	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/client-go/dynamic"
)

// detectExternalManager inspects labels and annotations for evidence that
// Helm, Argo CD, Flux, or another controller manages the resource. Returns
// the manager display name and a best-effort instance identifier; both empty
// when nothing was detected. Detection is deliberately conservative — radar and
// kubectl are not considered "external".
func detectExternalManager(obj *unstructured.Unstructured) (string, string) {
	if c := externalManagers(obj); len(c) > 0 {
		return c[0].name, c[0].instance
	}
	return "", ""
}

// externalManager is one manager an object's metadata names. claim: the
// metadata is a claim the GitOps manager answer weighs (gitops.Claims), so a
// caller reporting that answer needn't repeat it.
type externalManager struct {
	name, instance string
	claim          bool
}

// externalManagers returns every manager the metadata names, most specific
// first.
func externalManagers(obj *unstructured.Unstructured) []externalManager {
	if obj == nil {
		return nil
	}
	labels := obj.GetLabels()
	annotations := obj.GetAnnotations()
	var out []externalManager

	if v := labels["argocd.argoproj.io/instance"]; v != "" {
		out = append(out, externalManager{"Argo CD", v, true})
	}
	// A tracking ID copied from another object names that object's manager.
	if v := annotations["argocd.argoproj.io/tracking-id"]; v != "" && !copiedTrackingID(obj, v) {
		out = append(out, externalManager{"Argo CD", v, true})
	}

	if v := labels["kustomize.toolkit.fluxcd.io/name"]; v != "" {
		ns := labels["kustomize.toolkit.fluxcd.io/namespace"]
		instance := v
		if ns != "" {
			instance = ns + "/" + v
		}
		out = append(out, externalManager{"Flux (Kustomization)", instance, ns != ""})
	}
	if v := labels["helm.toolkit.fluxcd.io/name"]; v != "" {
		ns := labels["helm.toolkit.fluxcd.io/namespace"]
		instance := v
		if ns != "" {
			instance = ns + "/" + v
		}
		out = append(out, externalManager{"Flux (HelmRelease)", instance, ns != ""})
	}

	helmRelease := annotations["meta.helm.sh/release-name"]
	helmManaged := strings.EqualFold(labels["app.kubernetes.io/managed-by"], "Helm")
	if helmRelease != "" || helmManaged {
		instance := helmRelease
		if instance == "" {
			instance = labels["app.kubernetes.io/instance"]
		}
		out = append(out, externalManager{"Helm", instance, helmRelease != ""})
	}

	if mb := labels["app.kubernetes.io/managed-by"]; mb != "" && !strings.EqualFold(mb, "Helm") {
		low := strings.ToLower(mb)
		if low != "kubectl" && low != "radar" {
			out = append(out, externalManager{mb, labels["app.kubernetes.io/instance"], false})
		}
	}
	return out
}

// formatExternalManagerWarning words the warning as what the metadata says, not
// as a verdict: tracking metadata can be copied from another object or name a
// manager that no longer lists this one, and native Helm reconciles nothing
// between upgrades.
func formatExternalManagerWarning(manager, instance string) string {
	if manager == "" {
		return ""
	}
	frag := ""
	if instance != "" {
		frag = fmt.Sprintf(" (instance: %s)", instance)
	}
	if manager == "Helm" {
		return fmt.Sprintf("Its metadata says Helm installed it%s. The next helm upgrade or rollback of that release may overwrite direct edits; prefer changing the chart values.", frag)
	}
	return fmt.Sprintf("Its tracking metadata names %s%s. If that manager still tracks it, its next sync may revert direct edits; prefer changing the source of truth (Application spec / Git / operator CR).", manager, frag)
}

// podSpecPath returns the field path that contains the PodSpec for the given
// workload kind, or nil for kinds without an embedded PodSpec. Used by both
// the field-removal verification and the ConfigMap/Secret consumer lookup.
func podSpecPath(kind string) []string {
	switch kind {
	case "Deployment", "ReplicaSet", "StatefulSet", "DaemonSet", "Job":
		return []string{"spec", "template", "spec"}
	case "CronJob":
		return []string{"spec", "jobTemplate", "spec", "template", "spec"}
	case "Pod":
		return []string{"spec"}
	}
	return nil
}

// checkFieldRemoval verifies that fields the agent intended to remove are
// actually gone after apply. Server-side apply does not remove fields owned by
// other managers when those fields are merely omitted from the submission —
// the agent must explicitly set them to null. We check a focused set of
// high-impact paths under the PodSpec (probes, resources, scheduling).
//
// Returns one warning per field that survived the apply despite being omitted.
func checkFieldRemoval(submitted, pre, post *unstructured.Unstructured) []string {
	if submitted == nil || pre == nil || post == nil {
		return nil
	}
	specPath := podSpecPath(pre.GetKind())
	if len(specPath) == 0 {
		return nil
	}

	otherManagers := nonRadarManagers(post)

	var warnings []string

	for _, field := range []string{"nodeSelector", "affinity", "tolerations", "nodeName", "topologySpreadConstraints", "dnsPolicy", "dnsConfig", "hostAliases"} {
		path := append(append([]string{}, specPath...), field)
		if !hasPath(pre, path) || hasPath(submitted, path) {
			continue
		}
		if !hasPath(post, path) {
			continue
		}
		warnings = append(warnings, formatFieldRemovalWarning(strings.Join(path, "."), otherManagers))
	}

	warnings = append(warnings, checkContainerListRemoval("containers", specPath, submitted, pre, post, otherManagers)...)
	warnings = append(warnings, checkContainerListRemoval("initContainers", specPath, submitted, pre, post, otherManagers)...)

	return warnings
}

func checkContainerListRemoval(listField string, specPath []string, submitted, pre, post *unstructured.Unstructured, otherManagers []string) []string {
	preCs, _, _ := unstructured.NestedSlice(pre.Object, append(specPath, listField)...)
	submCs, _, _ := unstructured.NestedSlice(submitted.Object, append(specPath, listField)...)
	postCs, _, _ := unstructured.NestedSlice(post.Object, append(specPath, listField)...)
	submByName := containersByName(submCs)
	postByName := containersByName(postCs)

	var warnings []string
	for _, raw := range preCs {
		preC, ok := raw.(map[string]any)
		if !ok {
			continue
		}
		name, _ := preC["name"].(string)
		if name == "" {
			continue
		}
		submC, hasSubm := submByName[name]
		postC, hasPost := postByName[name]
		if !hasPost {
			continue
		}
		if !hasSubm {
			fieldRef := fmt.Sprintf("%s.%s[name=%s]", strings.Join(specPath, "."), listField, name)
			warnings = append(warnings, formatFieldRemovalWarning(fieldRef, otherManagers))
			continue
		}
		for _, field := range []string{"readinessProbe", "livenessProbe", "startupProbe", "resources", "env", "envFrom"} {
			if _, in := preC[field]; !in {
				continue
			}
			if _, in := submC[field]; in {
				continue
			}
			if _, in := postC[field]; !in {
				continue
			}
			fieldRef := fmt.Sprintf("%s.%s[name=%s].%s", strings.Join(specPath, "."), listField, name, field)
			warnings = append(warnings, formatFieldRemovalWarning(fieldRef, otherManagers))
		}
		for _, fieldRef := range retainedHostPortRefs(specPath, listField, name, preC, submC, postC) {
			warnings = append(warnings, formatFieldRemovalWarning(fieldRef, otherManagers))
		}
	}

	return warnings
}

func retainedHostPortRefs(specPath []string, listField, containerName string, preC, submC, postC map[string]any) []string {
	prePorts := hostPortByPort(preC)
	if len(prePorts) == 0 {
		return nil
	}
	submPorts := hostPortByPort(submC)
	postPorts := hostPortByPort(postC)
	var refs []string
	for key := range prePorts {
		if submPorts[key] {
			continue
		}
		if !postPorts[key] {
			continue
		}
		refs = append(refs, fmt.Sprintf("%s.%s[name=%s].ports[containerPort=%s].hostPort", strings.Join(specPath, "."), listField, containerName, key))
	}
	sort.Strings(refs)
	return refs
}

func hostPortByPort(container map[string]any) map[string]bool {
	rawPorts, ok := container["ports"].([]any)
	if !ok {
		return nil
	}
	out := map[string]bool{}
	for _, raw := range rawPorts {
		p, ok := raw.(map[string]any)
		if !ok {
			continue
		}
		if _, ok := p["hostPort"]; !ok {
			continue
		}
		containerPort, ok := scalarString(p["containerPort"])
		if !ok || containerPort == "" {
			continue
		}
		out[containerPort] = true
	}
	return out
}

func scalarString(v any) (string, bool) {
	switch t := v.(type) {
	case int:
		return fmt.Sprint(t), true
	case int32:
		return fmt.Sprint(t), true
	case int64:
		return fmt.Sprint(t), true
	case float64:
		return fmt.Sprintf("%.0f", t), true
	case string:
		return t, true
	default:
		return "", false
	}
}

func formatFieldRemovalWarning(field string, otherManagers []string) string {
	tail := "Server-side apply does not claim fields by omission. To actually remove the field, set its value to null in your YAML."
	if len(otherManagers) > 0 {
		tail = fmt.Sprintf("It is still owned by another field manager (%s). %s", strings.Join(otherManagers, ", "), tail)
	}
	return fmt.Sprintf("Field `%s` was omitted from your submission but is still present after apply. %s", field, tail)
}

func hasPath(obj *unstructured.Unstructured, path []string) bool {
	if obj == nil {
		return false
	}
	_, found, _ := unstructured.NestedFieldNoCopy(obj.Object, path...)
	return found
}

func containersByName(cs []any) map[string]map[string]any {
	out := make(map[string]map[string]any, len(cs))
	for _, c := range cs {
		m, ok := c.(map[string]any)
		if !ok {
			continue
		}
		name, _ := m["name"].(string)
		if name == "" {
			continue
		}
		out[name] = m
	}
	return out
}

// nonRadarManagers returns the names of field managers on the object other
// than radar itself, deduplicated and sorted for stable output. Used to tell
// the agent who else owns fields on the resource.
func nonRadarManagers(obj *unstructured.Unstructured) []string {
	if obj == nil {
		return nil
	}
	mfRaw, found, _ := unstructured.NestedSlice(obj.Object, "metadata", "managedFields")
	if !found {
		return nil
	}
	seen := map[string]struct{}{}
	for _, entry := range mfRaw {
		em, ok := entry.(map[string]any)
		if !ok {
			continue
		}
		name, _ := em["manager"].(string)
		if name == "" || strings.EqualFold(name, "radar") {
			continue
		}
		seen[name] = struct{}{}
	}
	if len(seen) == 0 {
		return nil
	}
	out := make([]string, 0, len(seen))
	for k := range seen {
		out = append(out, k)
	}
	sort.Strings(out)
	return out
}

// findConfigMapSecretConsumers returns the qualified names (Kind/Name) of
// workloads in the same namespace whose PodSpec references the given
// ConfigMap or Secret via env, envFrom, volumes, or projected volumes.
//
// partial is true when a workload kind could not be listed (e.g. RBAC denied
// the user the apply path impersonates, or the kind isn't in discovery). An
// empty result with partial=true means "scan incomplete", NOT "verified no
// consumers" — the caller must not present it as the latter.
func findConfigMapSecretConsumers(ctx context.Context, dynClient dynamic.Interface, discovery *ResourceDiscovery, namespace, refKind, refName string) (consumers []string, partial bool) {
	if dynClient == nil || discovery == nil || namespace == "" || refName == "" {
		return nil, false
	}
	if refKind != "ConfigMap" && refKind != "Secret" {
		return nil, false
	}

	for _, kind := range []string{"Deployment", "StatefulSet", "DaemonSet", "CronJob", "Job"} {
		gvr, ok := discovery.GetGVR(kind)
		if !ok {
			partial = true
			continue
		}
		list, err := dynClient.Resource(gvr).Namespace(namespace).List(ctx, metav1.ListOptions{})
		if err != nil {
			log.Printf("[k8s] apply_resource: consumer scan list %s in %s failed: %v", kind, namespace, err)
			partial = true
			continue
		}
		for i := range list.Items {
			item := &list.Items[i]
			if podSpecReferences(item, refKind, refName) {
				consumers = append(consumers, fmt.Sprintf("%s/%s", kind, item.GetName()))
			}
		}
	}
	sort.Strings(consumers)
	return consumers, partial
}

func podSpecReferences(obj *unstructured.Unstructured, refKind, refName string) bool {
	specPath := podSpecPath(obj.GetKind())
	if len(specPath) == 0 {
		return false
	}

	for _, group := range []string{"containers", "initContainers"} {
		cs, _, _ := unstructured.NestedSlice(obj.Object, append(specPath, group)...)
		for _, raw := range cs {
			c, ok := raw.(map[string]any)
			if !ok {
				continue
			}
			if containerReferences(c, refKind, refName) {
				return true
			}
		}
	}

	volumes, _, _ := unstructured.NestedSlice(obj.Object, append(specPath, "volumes")...)
	for _, raw := range volumes {
		v, ok := raw.(map[string]any)
		if !ok {
			continue
		}
		if volumeReferences(v, refKind, refName) {
			return true
		}
	}
	return false
}

func containerReferences(c map[string]any, refKind, refName string) bool {
	envFromKey := "configMapRef"
	envKey := "configMapKeyRef"
	if refKind == "Secret" {
		envFromKey = "secretRef"
		envKey = "secretKeyRef"
	}

	if envFrom, ok := c["envFrom"].([]any); ok {
		for _, raw := range envFrom {
			ef, ok := raw.(map[string]any)
			if !ok {
				continue
			}
			if ref, ok := ef[envFromKey].(map[string]any); ok {
				if n, _ := ref["name"].(string); n == refName {
					return true
				}
			}
		}
	}

	if env, ok := c["env"].([]any); ok {
		for _, raw := range env {
			ev, ok := raw.(map[string]any)
			if !ok {
				continue
			}
			vf, ok := ev["valueFrom"].(map[string]any)
			if !ok {
				continue
			}
			if ref, ok := vf[envKey].(map[string]any); ok {
				if n, _ := ref["name"].(string); n == refName {
					return true
				}
			}
		}
	}
	return false
}

func volumeReferences(v map[string]any, refKind, refName string) bool {
	if refKind == "ConfigMap" {
		if src, ok := v["configMap"].(map[string]any); ok {
			if n, _ := src["name"].(string); n == refName {
				return true
			}
		}
	}
	if refKind == "Secret" {
		if src, ok := v["secret"].(map[string]any); ok {
			if n, _ := src["secretName"].(string); n == refName {
				return true
			}
		}
	}

	if proj, ok := v["projected"].(map[string]any); ok {
		if sources, ok := proj["sources"].([]any); ok {
			for _, raw := range sources {
				sm, ok := raw.(map[string]any)
				if !ok {
					continue
				}
				if refKind == "ConfigMap" {
					if src, ok := sm["configMap"].(map[string]any); ok {
						if n, _ := src["name"].(string); n == refName {
							return true
						}
					}
				}
				if refKind == "Secret" {
					if src, ok := sm["secret"].(map[string]any); ok {
						if n, _ := src["name"].(string); n == refName {
							return true
						}
					}
				}
			}
		}
	}
	return false
}

func formatConsumerWarning(refKind, refName string, consumers []string, partial bool) string {
	if len(consumers) == 0 {
		if partial {
			return fmt.Sprintf("Could not fully enumerate workloads referencing `%s` (some list calls failed — likely RBAC). If you changed its contents, manually verify which workloads mount it and restart any that don't watch for changes.", refName)
		}
		return ""
	}
	preview := consumers
	tail := ""
	if len(preview) > 5 {
		preview = preview[:5]
		tail = fmt.Sprintf(" (+%d more)", len(consumers)-5)
	}
	incomplete := ""
	if partial {
		incomplete = " (consumer list may be incomplete — some list calls failed)"
	}
	return fmt.Sprintf("%s changes typically reach pod filesystems within ~60s, but most applications must detect and reload the new contents. %d workload(s) reference `%s`: %s%s%s. If those apps don't watch for changes, restart them (e.g., `manage_workload restart`) so they pick up the edit.", refKind, len(consumers), refName, strings.Join(preview, ", "), tail, incomplete)
}