package k8score

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/skyhook-io/radar/pkg/resourceid"

	appsv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	apiequality "k8s.io/apimachinery/pkg/api/equality"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/dynamic"
	"k8s.io/client-go/util/retry"
	"sigs.k8s.io/yaml"

	"github.com/skyhook-io/radar/pkg/rollouts"
)

// WorkloadRevision represents a single revision in a workload's rollout history.
type WorkloadRevision struct {
	Number    int64     `json:"number"`
	CreatedAt time.Time `json:"createdAt"`
	Image     string    `json:"image"` // primary container image
	IsCurrent bool      `json:"isCurrent"`
	Replicas  int64     `json:"replicas"`
	Template  string    `json:"template,omitempty"` // Pod template spec as YAML (for revision diff)
	// Rollouts only: mid-canary IsCurrent (rolling out) and IsStable (serving
	// traffic, what an abort reverts to) are different revisions.
	IsStable bool   `json:"isStable,omitempty"`
	PodHash  string `json:"podHash,omitempty"`
	// ReadyReplicas is this revision's own ready pods. Nil when the kind has no
	// per-revision count and the workload's pods may span revisions.
	ReadyReplicas *int64 `json:"readyReplicas,omitempty"`
	// Serving says whether this revision's pods are running and ready. Nil means
	// the API can't tell: a StatefulSet or DaemonSet mid-update reports no split
	// per revision. Rollouts leave it nil; IsStable is their signal.
	Serving *bool `json:"serving,omitempty"`
	// ChangeCause is the kubernetes.io/change-cause note on this revision, the
	// one record of why it exists when no Helm or GitOps source does.
	ChangeCause string `json:"changeCause,omitempty"`
}

// UpdateResourceOptions contains options for updating a resource.
type UpdateResourceOptions struct {
	Kind                    string
	Namespace               string
	Name                    string
	YAML                    string // YAML content to apply
	Force                   bool   // Force SSA field ownership conflicts (override Helm/Flux/Argo/kubectl)
	DryRun                  bool
	ExpectedResourceVersion string
}

type PreviewUpdateResourceResult struct {
	Object          *unstructured.Unstructured
	Live            *unstructured.Unstructured
	Submitted       *unstructured.Unstructured
	ResourceVersion string
	Warnings        []string
}

const radarOwnershipReclaimedWarning = "Radar reclaimed field ownership left by an earlier Radar write. No external manager ownership was overridden."

// DeleteResourceOptions contains options for deleting a resource.
type DeleteResourceOptions struct {
	Kind      string
	Group     string // API group, disambiguates kinds that collide across groups (e.g. Knative vs core Service)
	Namespace string
	Name      string
	Force     bool // Force delete with grace period 0
}

// ApplyResourceOptions contains options for creating or applying a resource.
type ApplyResourceOptions struct {
	YAML                    string // Raw YAML manifest
	Mode                    string // "apply" (server-side apply, default) or "create" (strict create)
	DryRun                  bool   // Validate without persisting
	NamespaceOverride       string // If set, overrides the namespace in the YAML
	Force                   bool   // Force SSA field ownership conflicts
	ExpectedResourceVersion string
	ExpectedResourceAbsent  bool
	UseCreateForAbsent      bool // Match a guarded create when the preflight confirms absence
}

// ApplyResourceResult contains the result of a create/apply operation.
type ApplyResourceResult struct {
	Name       string                     `json:"name"`
	Namespace  string                     `json:"namespace"`
	Kind       string                     `json:"kind"`
	APIVersion string                     `json:"apiVersion"`
	UID        types.UID                  `json:"uid,omitempty"`
	Created    bool                       `json:"created"` // true if newly created, false if updated
	Action     string                     `json:"action,omitempty"`
	Object     *unstructured.Unstructured `json:"-"`
	Previous   *unstructured.Unstructured `json:"-"`
	Submitted  *unstructured.Unstructured `json:"-"`

	// Warnings are advisory notes derived from the actual state of the cluster
	// (e.g., "its metadata says Helm installed it" or "field X you omitted is
	// still present after apply because manager Y owns it"). They never block
	// the apply — the apply succeeded if no error was returned. Treat each
	// entry as a self-contained sentence.
	Warnings []string `json:"warnings,omitempty"`
}

// WorkloadManager provides workload lifecycle operations using injected clients.
type WorkloadManager struct {
	dynClient dynamic.Interface
	discovery *ResourceDiscovery
}

// NewWorkloadManager creates a WorkloadManager with the given clients.
func NewWorkloadManager(dynClient dynamic.Interface, discovery *ResourceDiscovery) *WorkloadManager {
	return &WorkloadManager{dynClient: dynClient, discovery: discovery}
}

// UpdateResource updates a Kubernetes resource from YAML using server-side apply.
// SSA avoids the resourceVersion round-trip that PUT requires and matches
// kubectl apply --server-side / Lens semantics. Force=true takes ownership of
// fields the user is editing even if another field manager last wrote them.
func (m *WorkloadManager) UpdateResource(ctx context.Context, opts UpdateResourceOptions) (*unstructured.Unstructured, error) {
	obj, gvr, ri, err := m.prepareUpdateResource(opts)
	if err != nil {
		return nil, err
	}
	result, _, err := m.patchUpdateResource(ctx, opts, obj, gvr, ri)
	return result, err
}

func (m *WorkloadManager) PreviewUpdateResource(ctx context.Context, opts UpdateResourceOptions) (*PreviewUpdateResourceResult, error) {
	opts.DryRun = true
	obj, gvr, ri, err := m.prepareUpdateResource(opts)
	if err != nil {
		return nil, err
	}

	submitted := obj.DeepCopy()
	live, err := ri.Get(ctx, opts.Name, metav1.GetOptions{})
	if err != nil {
		return nil, fmt.Errorf("failed to read resource before preview: %w", err)
	}
	obj.SetResourceVersion(live.GetResourceVersion())
	predicted, reclaimedRadarOwnership, err := m.patchUpdateResource(ctx, opts, obj, gvr, ri)
	if err != nil {
		return nil, err
	}

	warnings := append(
		EnrichObjectWarnings(predicted),
		checkFieldRemoval(obj, live, predicted)...,
	)
	if reclaimedRadarOwnership {
		warnings = append([]string{radarOwnershipReclaimedWarning}, warnings...)
	}
	return &PreviewUpdateResourceResult{
		Object:          predicted,
		Live:            live,
		Submitted:       submitted,
		ResourceVersion: live.GetResourceVersion(),
		Warnings:        warnings,
	}, nil
}

func (m *WorkloadManager) prepareUpdateResource(opts UpdateResourceOptions) (*unstructured.Unstructured, schema.GroupVersionResource, dynamic.ResourceInterface, error) {
	if m.discovery == nil {
		return nil, schema.GroupVersionResource{}, nil, fmt.Errorf("resource discovery not initialized")
	}
	if m.dynClient == nil {
		return nil, schema.GroupVersionResource{}, nil, fmt.Errorf("dynamic client not initialized")
	}

	obj := &unstructured.Unstructured{}
	if err := yaml.Unmarshal([]byte(opts.YAML), &obj.Object); err != nil {
		return nil, schema.GroupVersionResource{}, nil, fmt.Errorf("invalid YAML: %w", err)
	}

	kindForLookup := opts.Kind
	if obj.GetKind() != "" {
		kindForLookup = obj.GetKind()
	}
	apiGroup := resourceid.GroupFromAPIVersion(obj.GetAPIVersion())
	var gvr schema.GroupVersionResource
	var ok bool
	if apiGroup != "" {
		gvr, ok = m.discovery.GetGVRWithGroup(kindForLookup, apiGroup)
		if !ok {
			if fallbackGVR, fallbackOK := m.discovery.GetGVR(kindForLookup); fallbackOK && fallbackGVR.Group == apiGroup {
				gvr, ok = fallbackGVR, true
			}
		}
	} else {
		gvr, ok = m.discovery.GetGVR(kindForLookup)
	}
	if !ok {
		return nil, schema.GroupVersionResource{}, nil, fmt.Errorf("unknown resource kind: %s", kindForLookup)
	}
	requestedGVR, requestedOK := m.discovery.GetGVRWithGroup(opts.Kind, apiGroup)
	if !requestedOK {
		requestedGVR, requestedOK = m.discovery.GetGVR(opts.Kind)
	}
	if !requestedOK {
		return nil, schema.GroupVersionResource{}, nil, fmt.Errorf("unknown resource kind: %s", opts.Kind)
	}
	if requestedGVR.Group != gvr.Group || requestedGVR.Resource != gvr.Resource {
		return nil, schema.GroupVersionResource{}, nil, fmt.Errorf("resource kind mismatch: expected %s, got %s", opts.Kind, kindForLookup)
	}

	if obj.GetName() != opts.Name {
		return nil, schema.GroupVersionResource{}, nil, fmt.Errorf("resource name mismatch: expected %s, got %s", opts.Name, obj.GetName())
	}
	if opts.Namespace != "" && obj.GetNamespace() != opts.Namespace {
		return nil, schema.GroupVersionResource{}, nil, fmt.Errorf("resource namespace mismatch: expected %s, got %s", opts.Namespace, obj.GetNamespace())
	}

	stripServerManagedFields(obj)
	var ri dynamic.ResourceInterface
	if opts.Namespace != "" {
		ri = m.dynClient.Resource(gvr).Namespace(opts.Namespace)
	} else {
		ri = m.dynClient.Resource(gvr)
	}
	return obj, gvr, ri, nil
}

func (m *WorkloadManager) patchUpdateResource(ctx context.Context, opts UpdateResourceOptions, obj *unstructured.Unstructured, gvr schema.GroupVersionResource, ri dynamic.ResourceInterface) (*unstructured.Unstructured, bool, error) {
	if opts.ExpectedResourceVersion != "" {
		live, err := ri.Get(ctx, opts.Name, metav1.GetOptions{})
		if err != nil {
			return nil, false, fmt.Errorf("failed to verify reviewed resource version: %w", err)
		}
		if live.GetResourceVersion() != opts.ExpectedResourceVersion {
			return nil, false, apierrors.NewConflict(
				gvr.GroupResource(),
				opts.Name,
				fmt.Errorf("resource changed after review; review the latest version before applying"),
			)
		}
		obj.SetResourceVersion(opts.ExpectedResourceVersion)
	}

	dryRun := []string(nil)
	if opts.DryRun {
		dryRun = []string{metav1.DryRunAll}
	}
	result, reclaimedRadarOwnership, err := patchWithRadarOwnershipReclaim(
		ctx,
		ri,
		opts.Name,
		obj,
		opts.Force,
		dryRun,
	)
	if err != nil {
		return nil, false, fmt.Errorf("failed to update resource: %w", err)
	}
	return result, reclaimedRadarOwnership, nil
}

// stripServerManagedFields removes metadata fields the apiserver owns. SSA
// rejects ownership claims on these, and editor round-trips often round-trip
// them back unchanged. status is intentionally NOT stripped: for resources
// with a status subresource the apiserver ignores it on /apply anyway, and
// for CRDs without a status subresource the user's edit IS the way to set it.
func stripServerManagedFields(obj *unstructured.Unstructured) {
	for _, f := range [][]string{
		{"metadata", "resourceVersion"},
		{"metadata", "managedFields"},
		{"metadata", "uid"},
		{"metadata", "generation"},
		{"metadata", "creationTimestamp"},
		{"metadata", "selfLink"},
	} {
		unstructured.RemoveNestedField(obj.Object, f...)
	}
}

// ApplyResource creates or updates a Kubernetes resource from YAML.
// In "apply" mode (default), uses server-side apply (idempotent create-or-update).
// In "create" mode, uses strict create (fails if resource exists).
func (m *WorkloadManager) ApplyResource(ctx context.Context, opts ApplyResourceOptions) (*ApplyResourceResult, error) {
	if m.discovery == nil {
		return nil, fmt.Errorf("resource discovery not initialized")
	}
	if m.dynClient == nil {
		return nil, fmt.Errorf("dynamic client not initialized")
	}

	obj := &unstructured.Unstructured{}
	if err := yaml.Unmarshal([]byte(opts.YAML), &obj.Object); err != nil {
		return nil, fmt.Errorf("invalid YAML: %w", err)
	}

	kind := obj.GetKind()
	if kind == "" {
		return nil, fmt.Errorf("YAML must include 'kind'")
	}
	apiVersion := obj.GetAPIVersion()
	if apiVersion == "" {
		return nil, fmt.Errorf("YAML must include 'apiVersion'")
	}
	name := obj.GetName()
	if name == "" {
		return nil, fmt.Errorf("YAML must include 'metadata.name'")
	}

	// Apply namespace override if provided
	if opts.NamespaceOverride != "" {
		obj.SetNamespace(opts.NamespaceOverride)
	}

	// Resolve GVR using group from apiVersion for disambiguation
	group := resourceid.GroupFromAPIVersion(apiVersion)

	var gvr schema.GroupVersionResource
	var ok bool
	if group != "" {
		gvr, ok = m.discovery.GetGVRWithGroup(kind, group)
	} else {
		gvr, ok = m.discovery.GetGVR(kind)
	}
	if !ok {
		return nil, fmt.Errorf("unknown resource kind: %s (apiVersion: %s)", kind, apiVersion)
	}

	ns := obj.GetNamespace()
	mode := opts.Mode
	if mode == "" {
		mode = "apply"
	}

	dryRun := []string{}
	if opts.DryRun {
		dryRun = []string{metav1.DryRunAll}
	}

	result := &ApplyResourceResult{
		Name:       name,
		Namespace:  ns,
		Kind:       kind,
		APIVersion: apiVersion,
		Submitted:  obj.DeepCopy(),
	}

	var client dynamic.ResourceInterface
	if ns != "" {
		client = m.dynClient.Resource(gvr).Namespace(ns)
	} else {
		client = m.dynClient.Resource(gvr)
	}
	// Pre-apply GET: feeds create/update classification, the external-manager warning, and the SSA
	// field-removal verification. Best-effort — a NotFound just means the
	// resource is being newly created, and other errors shouldn't block the
	// apply itself.
	var pre *unstructured.Unstructured
	var preGetErr error
	got, getErr := client.Get(ctx, name, metav1.GetOptions{})
	if getErr == nil {
		pre = got
		result.Action = "update"
	} else if apierrors.IsNotFound(getErr) {
		result.Action = "create"
	} else {
		preGetErr = getErr
		result.Action = "unknown"
		log.Printf("[k8s] apply_resource: pre-apply GET %s/%s/%s failed: %v", kind, ns, name, getErr)
	}
	result.Previous = pre
	if opts.ExpectedResourceAbsent {
		if preGetErr != nil {
			return result, fmt.Errorf("failed to verify reviewed resource absence: %w", preGetErr)
		}
		if pre != nil {
			return result, apierrors.NewConflict(
				gvr.GroupResource(),
				name,
				fmt.Errorf("resource was created after review; review the latest version before applying"),
			)
		}
	} else if opts.ExpectedResourceVersion != "" {
		if preGetErr != nil {
			return result, fmt.Errorf("failed to verify reviewed resource version: %w", preGetErr)
		}
		if pre == nil || pre.GetResourceVersion() != opts.ExpectedResourceVersion {
			return result, apierrors.NewConflict(
				gvr.GroupResource(),
				name,
				fmt.Errorf("resource changed after review; review the latest version before applying"),
			)
		}
		obj.SetResourceVersion(opts.ExpectedResourceVersion)
	}

	if mode == "create" {
		created, err := client.Create(ctx, obj, metav1.CreateOptions{
			DryRun:          dryRun,
			FieldManager:    "radar",
			FieldValidation: metav1.FieldValidationStrict,
		})
		if err != nil {
			return result, fmt.Errorf("failed to create resource: %w", err)
		}
		result.Created = true
		result.Action = "create"
		result.Object = created
		if !opts.DryRun {
			result.UID = created.GetUID()
		}
		m.populateApplyWarnings(ctx, result, obj, pre, created, ns, kind, name, opts.DryRun)
		return result, nil
	}

	if opts.ExpectedResourceAbsent || (opts.UseCreateForAbsent && pre == nil && preGetErr == nil) {
		// SSA has no must-not-exist precondition, so Create makes the reviewed
		// absence check atomic. Preview uses the same operation when absence is
		// observed so its admitted object matches the final write.
		created, err := client.Create(ctx, obj, metav1.CreateOptions{
			DryRun:          dryRun,
			FieldManager:    "radar",
			FieldValidation: metav1.FieldValidationStrict,
		})
		if err != nil {
			return result, fmt.Errorf("failed to apply reviewed create: %w", err)
		}
		result.Created = true
		result.Action = "create"
		result.Object = created
		if !opts.DryRun {
			result.UID = created.GetUID()
		}
		m.populateApplyWarnings(ctx, result, obj, pre, created, ns, kind, name, opts.DryRun)
		return result, nil
	}

	// Apply mode: server-side apply
	applied, reclaimedRadarOwnership, err := patchWithRadarOwnershipReclaim(
		ctx,
		client,
		name,
		obj,
		opts.Force,
		dryRun,
	)
	if err != nil {
		return result, fmt.Errorf("failed to apply resource: %w", err)
	}
	if reclaimedRadarOwnership {
		result.Warnings = append(result.Warnings, radarOwnershipReclaimedWarning)
	}

	// SSA results report "applied"; Action carries create/update classification.
	result.Created = false
	result.Object = applied
	if !opts.DryRun {
		result.UID = applied.GetUID()
	}

	// Post-apply GET feeds the state-derived warnings (run against what
	// actually landed) and the field-removal diff. Fetch it for creates too,
	// not just updates — an SSA apply that creates a resource still wants the
	// external-manager / terminating-namespace warnings (field-removal stays
	// gated on pre != nil below).
	var post *unstructured.Unstructured
	if opts.DryRun {
		post = applied
	} else {
		got, getErr := client.Get(ctx, name, metav1.GetOptions{})
		if getErr == nil {
			post = got
			result.Object = got
		} else {
			log.Printf("[k8s] apply_resource: post-apply GET %s/%s/%s failed: %v", kind, ns, name, getErr)
		}
	}

	// A non-NotFound pre-apply GET error means the object likely existed but we
	// couldn't read it, so field-retention can't be diffed (pre is nil). Tell the
	// agent the verification was skipped rather than letting silence imply a clean
	// apply — a partial manifest may have dropped fields without warning.
	if !opts.DryRun && pre == nil && preGetErr != nil {
		result.Warnings = append(result.Warnings, "Pre-apply GET failed; Radar could not compute field-retention warnings — a partial manifest may have silently dropped fields.")
	}

	m.populateApplyWarnings(ctx, result, obj, pre, post, ns, kind, name, opts.DryRun)
	return result, nil
}

// populateApplyWarnings appends advisory warnings to result.Warnings.
//
//   - State-derived warnings (external manager, deletionTimestamp, etc.) come
//     from the shared EnrichObjectWarnings; we run it against the admitted
//     object so the caller sees the resource the way any read of it would.
//   - Apply-specific checks (SSA field-removal verification, ConfigMap/Secret
//     consumer reload reminder) require knowledge of what was submitted vs.
//     what landed and so live here.
//
// Best-effort throughout — a failed check never fails the apply itself.
func (m *WorkloadManager) populateApplyWarnings(ctx context.Context, result *ApplyResourceResult, submitted, pre, post *unstructured.Unstructured, namespace, kind, name string, dryRun bool) {
	// State-derived: prefer the admitted object, then fall back to the prior live
	// object only when post-apply verification failed.
	target := post
	if target == nil {
		target = pre
	}
	result.Warnings = append(result.Warnings, EnrichObjectWarnings(target)...)

	if pre != nil && post != nil {
		result.Warnings = append(result.Warnings, checkFieldRemoval(submitted, pre, post)...)
	} else if !dryRun && pre != nil && post == nil {
		result.Warnings = append(result.Warnings, "Post-apply verification GET failed; Radar could not compute field-retention warnings from the live object.")
	}
	// The consumer-reload reminder only makes sense for a persisted edit — on a
	// dry run nothing landed, so the "restart consumers" advice would be
	// premature (and the namespace-wide LISTs wasted).
	if !dryRun && (kind == "ConfigMap" || kind == "Secret") && namespace != "" {
		consumers, partial := findConfigMapSecretConsumers(ctx, m.dynClient, m.discovery, namespace, kind, name)
		if w := formatConsumerWarning(kind, name, consumers, partial); w != "" {
			result.Warnings = append(result.Warnings, w)
		}
	}
}

func boolPtr(b bool) *bool { return &b }

func patchWithRadarOwnershipReclaim(
	ctx context.Context,
	client dynamic.ResourceInterface,
	name string,
	obj *unstructured.Unstructured,
	force bool,
	dryRun []string,
) (*unstructured.Unstructured, bool, error) {
	patch := func(body []byte, patchForce bool) (*unstructured.Unstructured, error) {
		return client.Patch(ctx, name, types.ApplyPatchType, body, metav1.PatchOptions{
			FieldManager:    "radar",
			FieldValidation: metav1.FieldValidationStrict,
			Force:           boolPtr(patchForce),
			DryRun:          dryRun,
		})
	}
	body, err := json.Marshal(obj.Object)
	if err != nil {
		return nil, false, fmt.Errorf("failed to marshal resource: %w", err)
	}
	applied, err := patch(body, force)
	if err == nil || force || !isOnlyRadarFieldManagerConflict(err) {
		return applied, false, err
	}

	retryObj := obj
	if retryObj.GetResourceVersion() == "" {
		live, getErr := client.Get(ctx, name, metav1.GetOptions{})
		if getErr != nil {
			return nil, false, err
		}
		retryObj = obj.DeepCopy()
		retryObj.SetResourceVersion(live.GetResourceVersion())
		body, err = json.Marshal(retryObj.Object)
		if err != nil {
			return nil, false, fmt.Errorf("failed to marshal resource: %w", err)
		}
		applied, err = patch(body, false)
		if err == nil || !isOnlyRadarFieldManagerConflict(err) {
			return applied, false, err
		}
	}

	applied, err = patch(body, true)
	return applied, err == nil, err
}

func isOnlyRadarFieldManagerConflict(err error) bool {
	var statusErr *apierrors.StatusError
	if !errors.As(err, &statusErr) || statusErr.ErrStatus.Details == nil {
		return false
	}
	causes := statusErr.ErrStatus.Details.Causes
	if len(causes) == 0 {
		return false
	}
	for _, cause := range causes {
		if cause.Type != metav1.CauseTypeFieldManagerConflict ||
			!strings.Contains(cause.Message, `conflict with "radar"`) {
			return false
		}
	}
	return true
}

// DeleteResource deletes a Kubernetes resource.
func (m *WorkloadManager) DeleteResource(ctx context.Context, opts DeleteResourceOptions) error {
	_, err := m.DeleteResourceWithResult(ctx, opts)
	return err
}

// DeleteResourceWithResult deletes a resource and observes asynchronous cleanup.
func (m *WorkloadManager) DeleteResourceWithResult(ctx context.Context, opts DeleteResourceOptions) (*DeleteResourceResult, error) {
	if m.discovery == nil {
		return nil, fmt.Errorf("resource discovery not initialized")
	}
	if m.dynClient == nil {
		return nil, fmt.Errorf("dynamic client not initialized")
	}

	gvr, ok := m.discovery.GetGVRWithGroup(opts.Kind, opts.Group)
	if !ok {
		return nil, apierrors.NewBadRequest(fmt.Sprintf("unknown resource kind: %s", opts.Kind))
	}

	var client dynamic.ResourceInterface = m.dynClient.Resource(gvr)
	if opts.Namespace != "" {
		client = m.dynClient.Resource(gvr).Namespace(opts.Namespace)
	}
	// A caller may have delete permission without get permission. Only observe
	// cleanup when a pre-delete read identifies the object by UID.
	var uid types.UID
	obj, readErr := client.Get(ctx, opts.Name, metav1.GetOptions{})
	if readErr == nil {
		uid = obj.GetUID()
	}

	// Background propagation avoids legacy defaults that orphan a Job's Pods.
	background := metav1.DeletePropagationBackground
	deleteOpts := metav1.DeleteOptions{PropagationPolicy: &background}
	if uid != "" {
		deleteOpts.Preconditions = &metav1.Preconditions{UID: &uid}
	}
	if readErr == nil {
		// Only GC finalizers record a policy to preserve; Background has no finalizer.
		// Omitting the policy otherwise can orphan a terminating Job's Pods.
		if _, ok := PendingDeletionPolicy(obj.GetFinalizers()); ok {
			deleteOpts.PropagationPolicy = nil
		}
	}

	if opts.Force && (readErr != nil || len(obj.GetFinalizers()) > 0) {
		metadata := map[string]any{"finalizers": nil}
		if readErr == nil {
			if obj.GetUID() != "" {
				metadata["uid"] = obj.GetUID()
			}
			if obj.GetResourceVersion() != "" {
				metadata["resourceVersion"] = obj.GetResourceVersion()
			}
		}
		finalizerPatch, _ := json.Marshal(map[string]any{"metadata": metadata})
		var patchErr error
		if opts.Namespace != "" {
			_, patchErr = m.dynClient.Resource(gvr).Namespace(opts.Namespace).Patch(ctx, opts.Name, types.MergePatchType, finalizerPatch, metav1.PatchOptions{})
		} else {
			_, patchErr = m.dynClient.Resource(gvr).Patch(ctx, opts.Name, types.MergePatchType, finalizerPatch, metav1.PatchOptions{})
		}
		if patchErr != nil && !apierrors.IsNotFound(patchErr) {
			if apierrors.IsForbidden(patchErr) {
				return nil, fmt.Errorf("force delete requires patch permission to strip finalizers: %w", patchErr)
			}
			return nil, fmt.Errorf("force delete could not strip finalizers: %w", patchErr)
		}
	}

	if opts.Force {
		deleteOpts.PropagationPolicy = &background
		gracePeriod := int64(0)
		deleteOpts.GracePeriodSeconds = &gracePeriod
	}

	var err error
	if opts.Namespace != "" {
		err = m.dynClient.Resource(gvr).Namespace(opts.Namespace).Delete(ctx, opts.Name, deleteOpts)
	} else {
		err = m.dynClient.Resource(gvr).Delete(ctx, opts.Name, deleteOpts)
	}
	if err != nil {
		if opts.Force && apierrors.IsNotFound(err) {
			return &DeleteResourceResult{}, nil
		}
		return nil, fmt.Errorf("failed to delete resource: %w", err)
	}

	if uid == "" {
		result := &DeleteResourceResult{ObservationError: "Deletion target UID is unavailable; cleanup was not observed."}
		if readErr != nil && !apierrors.IsNotFound(readErr) {
			result.ObservationError = readErr.Error()
		}
		return result, nil
	}

	return ObserveResourceDeletion(ctx, client, opts.Name, uid), nil
}

// TriggerCronJob creates a Job from a CronJob.
func (m *WorkloadManager) TriggerCronJob(ctx context.Context, namespace, name string) (*unstructured.Unstructured, error) {
	if m.dynClient == nil {
		return nil, fmt.Errorf("dynamic client not initialized")
	}
	if m.discovery == nil {
		return nil, fmt.Errorf("resource discovery not initialized")
	}

	cronJobGVR, ok := m.discovery.GetGVR("cronjobs")
	if !ok {
		return nil, fmt.Errorf("cronjobs resource not found")
	}

	cronJob, err := m.dynClient.Resource(cronJobGVR).Namespace(namespace).Get(ctx, name, metav1.GetOptions{})
	if err != nil {
		return nil, fmt.Errorf("failed to get cronjob: %w", err)
	}

	jobTemplate, found, err := unstructured.NestedMap(cronJob.Object, "spec", "jobTemplate")
	if err != nil {
		return nil, fmt.Errorf("failed to get job template from cronjob %s/%s: %w", namespace, name, err)
	}
	if !found {
		return nil, fmt.Errorf("cronjob %s/%s has no spec.jobTemplate", namespace, name)
	}

	jobName := fmt.Sprintf("%s-manual-%d", name, time.Now().Unix())
	job := &unstructured.Unstructured{
		Object: map[string]any{
			"apiVersion": "batch/v1",
			"kind":       "Job",
			"metadata": map[string]any{
				"name":      jobName,
				"namespace": namespace,
				"annotations": map[string]any{
					"cronjob.kubernetes.io/instantiate": "manual",
				},
				"ownerReferences": []any{
					map[string]any{
						"apiVersion":         cronJob.GetAPIVersion(),
						"kind":               cronJob.GetKind(),
						"name":               cronJob.GetName(),
						"uid":                string(cronJob.GetUID()),
						"controller":         true,
						"blockOwnerDeletion": true,
					},
				},
			},
		},
	}

	if spec, found, _ := unstructured.NestedMap(jobTemplate, "spec"); found {
		if err := unstructured.SetNestedMap(job.Object, spec, "spec"); err != nil {
			return nil, fmt.Errorf("failed to set job spec: %w", err)
		}
	}
	if labels, found, _ := unstructured.NestedStringMap(jobTemplate, "metadata", "labels"); found {
		if err := unstructured.SetNestedStringMap(job.Object, labels, "metadata", "labels"); err != nil {
			return nil, fmt.Errorf("failed to set job labels: %w", err)
		}
	}

	jobGVR, ok := m.discovery.GetGVR("jobs")
	if !ok {
		return nil, fmt.Errorf("jobs resource not found")
	}

	result, err := m.dynClient.Resource(jobGVR).Namespace(namespace).Create(ctx, job, metav1.CreateOptions{})
	if err != nil {
		return nil, fmt.Errorf("failed to create job: %w", err)
	}

	return result, nil
}

// SetCronJobSuspend sets the suspend field on a CronJob.
func (m *WorkloadManager) SetCronJobSuspend(ctx context.Context, namespace, name string, suspend bool) error {
	if m.dynClient == nil {
		return fmt.Errorf("dynamic client not initialized")
	}
	if m.discovery == nil {
		return fmt.Errorf("resource discovery not initialized")
	}

	cronJobGVR, ok := m.discovery.GetGVR("cronjobs")
	if !ok {
		return fmt.Errorf("cronjobs resource not found")
	}

	patch := fmt.Sprintf(`{"spec":{"suspend":%t}}`, suspend)
	_, err := m.dynClient.Resource(cronJobGVR).Namespace(namespace).Patch(
		ctx, name, types.MergePatchType, []byte(patch), metav1.PatchOptions{},
	)
	if err != nil {
		return fmt.Errorf("failed to patch cronjob: %w", err)
	}

	return nil
}

// RestartWorkload performs a rolling restart on a Deployment, StatefulSet, or DaemonSet.
func (m *WorkloadManager) RestartWorkload(ctx context.Context, kind, namespace, name string) error {
	if m.dynClient == nil {
		return fmt.Errorf("dynamic client not initialized")
	}
	normalizedKind := NormalizeWorkloadKind(strings.ToLower(kind))
	// A Rollout's restart is spec.restartAt; the pod-template annotation below
	// would change the template hash and re-run every canary step instead.
	if normalizedKind == "rollouts" {
		_, err := rollouts.Restart(ctx, m.dynClient, namespace, name)
		return err
	}

	if m.discovery == nil {
		return fmt.Errorf("resource discovery not initialized")
	}

	gvr, ok := m.discovery.GetGVR(kind)
	if !ok {
		return fmt.Errorf("unknown resource kind: %s", kind)
	}

	if normalizedKind == "deployments" {
		deployment, err := m.dynClient.Resource(gvr).Namespace(namespace).Get(ctx, name, metav1.GetOptions{})
		if err != nil {
			return fmt.Errorf("failed to restart workload: %w", err)
		}
		if paused, _, _ := unstructured.NestedBool(deployment.Object, "spec", "paused"); paused {
			return fmt.Errorf("Deployment %s/%s: %w", namespace, name, ErrRestartPaused)
		}
	}

	restartTime := time.Now().Format(time.RFC3339)
	patch := fmt.Sprintf(`{"spec":{"template":{"metadata":{"annotations":{"kubectl.kubernetes.io/restartedAt":"%s"}}}}}`, restartTime)

	_, err := m.dynClient.Resource(gvr).Namespace(namespace).Patch(
		ctx, name, types.MergePatchType, []byte(patch), metav1.PatchOptions{},
	)
	if err != nil {
		return fmt.Errorf("failed to restart workload: %w", err)
	}

	return nil
}

// SetDeploymentPaused sets spec.paused on a Deployment, as kubectl rollout pause
// and resume do. The API server treats a patch to the value already set as a
// no-op, so a repeated call succeeds and changes nothing without a prior read.
func (m *WorkloadManager) SetDeploymentPaused(ctx context.Context, namespace, name string, paused bool) error {
	if m.dynClient == nil {
		return fmt.Errorf("dynamic client not initialized")
	}
	if m.discovery == nil {
		return fmt.Errorf("resource discovery not initialized")
	}

	deploymentGVR, ok := m.discovery.GetGVR("deployments")
	if !ok {
		return fmt.Errorf("deployments resource not found")
	}

	patch := fmt.Sprintf(`{"spec":{"paused":%t}}`, paused)
	if _, err := m.dynClient.Resource(deploymentGVR).Namespace(namespace).Patch(
		ctx, name, types.MergePatchType, []byte(patch), metav1.PatchOptions{},
	); err != nil {
		return fmt.Errorf("failed to patch deployment: %w", err)
	}

	return nil
}

// ScaleWorkload scales a Deployment, StatefulSet, or Rollout to the specified replica count.
func (m *WorkloadManager) ScaleWorkload(ctx context.Context, kind, namespace, name string, replicas int32) error {
	if m.dynClient == nil {
		return fmt.Errorf("dynamic client not initialized")
	}

	normalizedKind := NormalizeWorkloadKind(kind)
	var gvr schema.GroupVersionResource
	switch normalizedKind {
	case "rollouts":
		// Fixed, not discovered: the CRD may not be in discovery's cache yet.
		gvr = rollouts.GVR
	case "deployments", "statefulsets":
		if m.discovery == nil {
			return fmt.Errorf("resource discovery not initialized")
		}
		found, ok := m.discovery.GetGVR(normalizedKind)
		if !ok {
			return fmt.Errorf("unknown resource kind: %s", kind)
		}
		gvr = found
	default:
		return fmt.Errorf("scaling not supported for %s (only deployments, statefulsets, and rollouts)", kind)
	}

	patch := fmt.Sprintf(`{"spec":{"replicas":%d}}`, replicas)
	if _, err := m.dynClient.Resource(gvr).Namespace(namespace).Patch(
		ctx, name, types.MergePatchType, []byte(patch), metav1.PatchOptions{},
	); err != nil {
		return fmt.Errorf("failed to scale workload: %w", err)
	}

	return nil
}

// ScaleWorkloadDirect scales a Deployment or StatefulSet without requiring a WorkloadManager.
// It accepts a dynamic.Interface directly, for use by callers that don't have a full WorkloadManager.
func ScaleWorkloadDirect(ctx context.Context, dynClient dynamic.Interface, kind, namespace, name string, replicas int32) error {
	if dynClient == nil {
		return fmt.Errorf("dynamic client must not be nil")
	}

	normalizedKind := NormalizeWorkloadKind(kind)

	var gvr schema.GroupVersionResource
	switch normalizedKind {
	case "deployments":
		gvr = schema.GroupVersionResource{Group: "apps", Version: "v1", Resource: "deployments"}
	case "statefulsets":
		gvr = schema.GroupVersionResource{Group: "apps", Version: "v1", Resource: "statefulsets"}
	default:
		return fmt.Errorf("scaling not supported for %s (only deployments and statefulsets)", kind)
	}

	patch := fmt.Sprintf(`{"spec":{"replicas":%d}}`, replicas)
	_, err := dynClient.Resource(gvr).Namespace(namespace).Patch(
		ctx, name, types.MergePatchType, []byte(patch), metav1.PatchOptions{},
	)
	if err != nil {
		return fmt.Errorf("failed to scale workload: %w", err)
	}

	return nil
}

// ListWorkloadRevisions returns the revision history for a Deployment, StatefulSet, DaemonSet, or Rollout.
func (m *WorkloadManager) ListWorkloadRevisions(ctx context.Context, kind, namespace, name string) ([]WorkloadRevision, error) {
	if m.dynClient == nil {
		return nil, fmt.Errorf("dynamic client not initialized")
	}

	normalizedKind := NormalizeWorkloadKind(kind)

	// pkg/rollouts carries its own GVR, so this path works before discovery has
	// cached the Rollout CRD.
	if normalizedKind == "rollouts" {
		return m.listRolloutRevisions(ctx, namespace, name)
	}

	if m.discovery == nil {
		return nil, fmt.Errorf("resource discovery not initialized")
	}

	workloadGVR, ok := m.discovery.GetGVR(normalizedKind)
	if !ok {
		return nil, fmt.Errorf("unknown resource kind: %s", kind)
	}

	workload, err := m.dynClient.Resource(workloadGVR).Namespace(namespace).Get(ctx, name, metav1.GetOptions{})
	if err != nil {
		return nil, fmt.Errorf("failed to get workload: %w", err)
	}
	workloadUID := string(workload.GetUID())

	switch normalizedKind {
	case "deployments":
		return m.listDeploymentRevisions(ctx, namespace, name, workloadUID)
	case "statefulsets", "daemonsets":
		revisions, err := m.listControllerRevisions(ctx, namespace, name, workloadUID)
		if err != nil {
			return nil, err
		}
		applyControllerRevisionServing(revisions, workload, normalizedKind)
		return revisions, nil
	default:
		return nil, fmt.Errorf("revision history not supported for %s", kind)
	}
}

func (m *WorkloadManager) listRolloutRevisions(ctx context.Context, namespace, name string) ([]WorkloadRevision, error) {
	revisions, err := rollouts.ListRevisions(ctx, m.dynClient, namespace, name)
	if err != nil {
		return nil, err
	}
	out := make([]WorkloadRevision, 0, len(revisions))
	for _, r := range revisions {
		readyReplicas := r.ReadyReplicas
		out = append(out, WorkloadRevision{
			Number:        r.Number,
			CreatedAt:     r.CreatedAt,
			Image:         r.Image,
			IsCurrent:     r.IsCurrent,
			IsStable:      r.IsStable,
			PodHash:       r.PodHash,
			Replicas:      r.Replicas,
			Template:      r.Template,
			ReadyReplicas: &readyReplicas,
		})
	}
	return out, nil
}

func (m *WorkloadManager) listDeploymentRevisions(ctx context.Context, namespace, name, workloadUID string) ([]WorkloadRevision, error) {
	rsGVR, ok := m.discovery.GetGVR("replicasets")
	if !ok {
		return nil, fmt.Errorf("replicasets resource not found")
	}

	rsList, err := m.dynClient.Resource(rsGVR).Namespace(namespace).List(ctx, metav1.ListOptions{})
	if err != nil {
		return nil, fmt.Errorf("failed to list replicasets: %w", err)
	}

	return BuildDeploymentRevisions(rsList.Items, workloadUID), nil
}

// BuildDeploymentRevisions builds WorkloadRevision[] from unstructured ReplicaSets owned by a Deployment.
// workloadUID filters by UID-based ownership; pass empty string to skip UID-based filter.
func BuildDeploymentRevisions(rsList []unstructured.Unstructured, workloadUID string) []WorkloadRevision {
	var revisions []WorkloadRevision
	var maxRevision int64

	for _, rs := range rsList {
		if workloadUID != "" {
			owned := false
			for _, ref := range rs.GetOwnerReferences() {
				if string(ref.UID) == workloadUID {
					owned = true
					break
				}
			}
			if !owned {
				continue
			}
		}

		revStr, ok := rs.GetAnnotations()["deployment.kubernetes.io/revision"]
		if !ok {
			continue
		}
		revNum, err := strconv.ParseInt(revStr, 10, 64)
		if err != nil {
			continue
		}

		image := extractContainerImage(rs.Object)
		replicas, _, _ := unstructured.NestedInt64(rs.Object, "spec", "replicas")
		readyReplicas, _, _ := unstructured.NestedInt64(rs.Object, "status", "readyReplicas")
		serving := readyReplicas > 0

		var templateStr string
		if template, found, _ := unstructured.NestedMap(rs.Object, "spec", "template"); found && template != nil {
			if templateYAML, err := yaml.Marshal(template); err == nil {
				templateStr = string(templateYAML)
			}
		}

		if revNum > maxRevision {
			maxRevision = revNum
		}

		revisions = append(revisions, WorkloadRevision{
			Number:        revNum,
			CreatedAt:     rs.GetCreationTimestamp().Time,
			Image:         image,
			Replicas:      replicas,
			Template:      templateStr,
			PodHash:       rs.GetLabels()["pod-template-hash"],
			ReadyReplicas: &readyReplicas,
			Serving:       &serving,
			ChangeCause:   rs.GetAnnotations()["kubernetes.io/change-cause"],
		})
	}

	for i := range revisions {
		if revisions[i].Number == maxRevision {
			revisions[i].IsCurrent = true
		}
	}

	sort.Slice(revisions, func(i, j int) bool { return revisions[i].Number > revisions[j].Number })

	return revisions
}

func CurrentDeploymentRevisionPodHashes(replicaSets []*appsv1.ReplicaSet) map[types.UID]string {
	type revisionHash struct {
		revision int64
		hash     string
	}
	current := make(map[types.UID]revisionHash)
	for _, replicaSet := range replicaSets {
		revision, err := strconv.ParseInt(replicaSet.Annotations["deployment.kubernetes.io/revision"], 10, 64)
		if err != nil {
			continue
		}
		for _, owner := range replicaSet.OwnerReferences {
			if owner.Kind != "Deployment" || owner.UID == "" {
				continue
			}
			if existing, ok := current[owner.UID]; !ok || revision > existing.revision {
				current[owner.UID] = revisionHash{revision: revision, hash: replicaSet.Labels[appsv1.DefaultDeploymentUniqueLabelKey]}
			}
			break
		}
	}
	hashes := make(map[types.UID]string, len(current))
	for uid, revision := range current {
		if revision.hash != "" {
			hashes[uid] = revision.hash
		}
	}
	return hashes
}

func CurrentControllerRevisionPodHashes(revisions []unstructured.Unstructured) map[types.UID]string {
	type revisionHash struct {
		revision int64
		hash     string
	}
	current := make(map[types.UID]revisionHash)
	for i := range revisions {
		revision := &revisions[i]
		number, found, _ := unstructured.NestedInt64(revision.Object, "revision")
		if !found {
			continue
		}
		for _, owner := range revision.GetOwnerReferences() {
			if (owner.Kind != "StatefulSet" && owner.Kind != "DaemonSet") || owner.UID == "" {
				continue
			}
			if existing, ok := current[owner.UID]; !ok || number > existing.revision {
				current[owner.UID] = revisionHash{revision: number, hash: revision.GetLabels()[appsv1.ControllerRevisionHashLabelKey]}
			}
			break
		}
	}
	hashes := make(map[types.UID]string, len(current))
	for uid, revision := range current {
		if revision.hash != "" {
			hashes[uid] = revision.hash
		}
	}
	return hashes
}

func (m *WorkloadManager) listControllerRevisions(ctx context.Context, namespace, name, workloadUID string) ([]WorkloadRevision, error) {
	crGVR, ok := m.discovery.GetGVR("controllerrevisions")
	if !ok {
		return nil, fmt.Errorf("controllerrevisions resource not found")
	}

	crList, err := m.dynClient.Resource(crGVR).Namespace(namespace).List(ctx, metav1.ListOptions{})
	if err != nil {
		return nil, fmt.Errorf("failed to list controllerrevisions: %w", err)
	}

	return BuildControllerRevisions(crList.Items, workloadUID), nil
}

// BuildControllerRevisions builds WorkloadRevision[] from unstructured ControllerRevisions owned by a StatefulSet or DaemonSet.
// workloadUID filters by UID-based ownership; pass empty string to skip UID-based filter.
func BuildControllerRevisions(crList []unstructured.Unstructured, workloadUID string) []WorkloadRevision {
	var revisions []WorkloadRevision
	var maxRevision int64

	for _, cr := range crList {
		ownerKind := ""
		if workloadUID != "" {
			owned := false
			for _, ref := range cr.GetOwnerReferences() {
				if string(ref.UID) == workloadUID {
					owned = true
					ownerKind = ref.Kind
					break
				}
			}
			if !owned {
				continue
			}
		}

		revNum, found, err := unstructured.NestedInt64(cr.Object, "revision")
		if err != nil || !found {
			continue
		}

		image := extractContainerImageFromData(cr.Object)

		var templateStr string
		if data, found, _ := unstructured.NestedMap(cr.Object, "data"); found && data != nil {
			if template, tFound, _ := unstructured.NestedMap(data, "spec", "template"); tFound && template != nil {
				if templateYAML, err := yaml.Marshal(template); err == nil {
					templateStr = string(templateYAML)
				}
			} else {
				if dataYAML, err := yaml.Marshal(data); err == nil {
					templateStr = string(dataYAML)
				}
			}
		}

		if revNum > maxRevision {
			maxRevision = revNum
		}

		podHash := cr.GetLabels()[appsv1.ControllerRevisionHashLabelKey]
		if ownerKind == "StatefulSet" {
			podHash = cr.GetName()
		}
		revisions = append(revisions, WorkloadRevision{
			Number:    revNum,
			CreatedAt: cr.GetCreationTimestamp().Time,
			Image:     image,
			Template:  templateStr,
			PodHash:   podHash,
		})
	}

	for i := range revisions {
		if revisions[i].Number == maxRevision {
			revisions[i].IsCurrent = true
		}
	}

	sort.Slice(revisions, func(i, j int) bool { return revisions[i].Number > revisions[j].Number })

	return revisions
}

// applyControllerRevisionServing fills the serving signal when the workload's
// status proves every pod is on one revision. Mid-update there is no per-revision
// pod count to read, so the fields stay nil rather than guessing.
func applyControllerRevisionServing(revisions []WorkloadRevision, workload *unstructured.Unstructured, normalizedKind string) {
	generation := workload.GetGeneration()
	observed, _, _ := unstructured.NestedInt64(workload.Object, "status", "observedGeneration")
	if observed < generation {
		return
	}

	settled := -1
	var ready int64
	switch normalizedKind {
	case "statefulsets":
		updateRevision, _, _ := unstructured.NestedString(workload.Object, "status", "updateRevision")
		currentRevision, _, _ := unstructured.NestedString(workload.Object, "status", "currentRevision")
		if updateRevision == "" || updateRevision != currentRevision {
			return
		}
		// OnDelete can leave pods on a reverted revision while both names agree,
		// and those pods are the only ones this count leaves out.
		replicas, _, _ := unstructured.NestedInt64(workload.Object, "status", "replicas")
		onRevision, _, _ := unstructured.NestedInt64(workload.Object, "status", "currentReplicas")
		if onRevision != replicas {
			return
		}
		ready, _, _ = unstructured.NestedInt64(workload.Object, "status", "readyReplicas")
		for i := range revisions {
			if revisions[i].PodHash == updateRevision {
				settled = i
			}
		}
	case "daemonsets":
		desired, _, _ := unstructured.NestedInt64(workload.Object, "status", "desiredNumberScheduled")
		updated, _, _ := unstructured.NestedInt64(workload.Object, "status", "updatedNumberScheduled")
		// Pods on nodes the DaemonSet no longer targets count in neither figure,
		// and they can be on an older revision.
		misscheduled, _, _ := unstructured.NestedInt64(workload.Object, "status", "numberMisscheduled")
		if updated != desired || misscheduled != 0 {
			return
		}
		ready, _, _ = unstructured.NestedInt64(workload.Object, "status", "numberReady")
		for i := range revisions {
			if revisions[i].IsCurrent {
				settled = i
			}
		}
	}
	if settled < 0 {
		return
	}

	for i := range revisions {
		serving := i == settled && ready > 0
		revisions[i].Serving = &serving
		var count int64
		if i == settled {
			count = ready
		}
		revisions[i].ReadyReplicas = &count
	}
}

// RollbackWorkload rolls back a Deployment, StatefulSet, DaemonSet, or Rollout to a specific revision.
func (m *WorkloadManager) RollbackWorkload(ctx context.Context, kind, namespace, name string, revision int64) error {
	if m.dynClient == nil {
		return fmt.Errorf("dynamic client not initialized")
	}

	normalizedKind := NormalizeWorkloadKind(kind)

	if normalizedKind == "rollouts" {
		_, err := rollouts.Undo(ctx, m.dynClient, namespace, name, revision)
		return err
	}

	if m.discovery == nil {
		return fmt.Errorf("resource discovery not initialized")
	}

	workloadGVR, ok := m.discovery.GetGVR(normalizedKind)
	if !ok {
		return fmt.Errorf("unknown resource kind: %s", kind)
	}

	switch normalizedKind {
	case "deployments", "statefulsets", "daemonsets":
	default:
		return fmt.Errorf("rollback not supported for %s", kind)
	}

	// The patch pins the resourceVersion that was read, and controller status
	// writes bump it, so a rollback during an active rollout can conflict without
	// any change to the template. Each retry re-reads and re-compares.
	return retry.RetryOnConflict(retry.DefaultRetry, func() error {
		workload, err := m.dynClient.Resource(workloadGVR).Namespace(namespace).Get(ctx, name, metav1.GetOptions{})
		if err != nil {
			return fmt.Errorf("failed to get workload: %w", err)
		}
		if normalizedKind == "deployments" {
			return m.rollbackDeployment(ctx, workload, revision)
		}
		return m.rollbackControllerRevision(ctx, normalizedKind, workload, revision)
	})
}

// ErrWorkloadPaused: a paused Deployment would hold a restored template without
// rolling it out, so a rollback that reports success would change nothing.
// kubectl rollout undo refuses for the same reason.
var ErrWorkloadPaused = errors.New("the Deployment is paused, so a rollback would not roll out; resume it, then roll back")

// ErrRestartPaused: a paused Deployment would record the restart without
// rolling it out, and then roll it out on the next resume. kubectl rollout
// restart refuses for the same reason.
var ErrRestartPaused = errors.New("the Deployment is paused, so a restart would not roll out; resume it, then restart")

// deploymentRollbackKeptAnnotations are the Deployment annotations a rollback keeps
// from the live object; every other annotation is restored from the target
// ReplicaSet. Same set as kubectl rollout undo's annotationsToSkip
// (k8s.io/kubectl/pkg/polymorphichelpers/rollback.go).
var deploymentRollbackKeptAnnotations = map[string]bool{
	corev1.LastAppliedConfigAnnotation:          true,
	"deployment.kubernetes.io/revision":         true,
	"deployment.kubernetes.io/revision-history": true,
	"deployment.kubernetes.io/desired-replicas": true,
	"deployment.kubernetes.io/max-replicas":     true,
	appsv1.DeprecatedRollbackTo:                 true,
}

// rollbackDeployment restores a ReplicaSet's pod template the way kubectl rollout
// undo does: the whole spec.template is replaced (a merge patch would keep fields
// the target revision lacks), pod-template-hash is dropped (the controller adds
// it to the ReplicaSet, never to the Deployment), and annotations come from the
// ReplicaSet except the controller's bookkeeping. The patch carries the
// resourceVersion that was compared, so a concurrent write fails with a conflict
// instead of being overwritten.
func (m *WorkloadManager) rollbackDeployment(ctx context.Context, deployment *unstructured.Unstructured, revision int64) error {
	namespace, name := deployment.GetNamespace(), deployment.GetName()
	if paused, _, _ := unstructured.NestedBool(deployment.Object, "spec", "paused"); paused {
		return fmt.Errorf("Deployment %s/%s: %w", namespace, name, ErrWorkloadPaused)
	}

	rsGVR, ok := m.discovery.GetGVR("replicasets")
	if !ok {
		return fmt.Errorf("replicasets resource not found")
	}
	rsList, err := m.dynClient.Resource(rsGVR).Namespace(namespace).List(ctx, metav1.ListOptions{})
	if err != nil {
		return fmt.Errorf("failed to list replicasets: %w", err)
	}

	var targetRS *unstructured.Unstructured
	for i := range rsList.Items {
		rs := &rsList.Items[i]
		if !ownedBy(rs, deployment.GetUID()) {
			continue
		}
		revNum, err := strconv.ParseInt(rs.GetAnnotations()["deployment.kubernetes.io/revision"], 10, 64)
		if err == nil && revNum == revision {
			targetRS = rs
			break
		}
	}
	if targetRS == nil {
		return fmt.Errorf("revision %d not found", revision)
	}

	template, found, err := unstructured.NestedMap(targetRS.Object, "spec", "template")
	if err != nil || !found {
		return fmt.Errorf("failed to extract pod template from revision %d", revision)
	}
	if labels, found, _ := unstructured.NestedStringMap(template, "metadata", "labels"); found {
		delete(labels, appsv1.DefaultDeploymentUniqueLabelKey)
		if len(labels) == 0 {
			unstructured.RemoveNestedField(template, "metadata", "labels")
		} else if err := unstructured.SetNestedStringMap(template, labels, "metadata", "labels"); err != nil {
			return fmt.Errorf("failed to strip pod-template-hash from revision %d: %w", revision, err)
		}
	}

	live, _, _ := unstructured.NestedMap(deployment.Object, "spec", "template")
	if same, err := podTemplatesEqual(live, template, true); err != nil {
		return fmt.Errorf("failed to compare revision %d with the current template: %w", revision, err)
	} else if same {
		return fmt.Errorf("revision %d of Deployment %s/%s: %w", revision, namespace, name, rollouts.ErrTemplateUnchanged)
	}

	annotations := map[string]string{}
	for k, v := range deployment.GetAnnotations() {
		if deploymentRollbackKeptAnnotations[k] {
			annotations[k] = v
		}
	}
	for k, v := range targetRS.GetAnnotations() {
		if !deploymentRollbackKeptAnnotations[k] {
			annotations[k] = v
		}
	}

	patch, err := json.Marshal([]map[string]any{
		{"op": "replace", "path": "/metadata/resourceVersion", "value": deployment.GetResourceVersion()},
		{"op": "replace", "path": "/spec/template", "value": template},
		{"op": "add", "path": "/metadata/annotations", "value": annotations},
	})
	if err != nil {
		return fmt.Errorf("failed to build rollback patch: %w", err)
	}

	deployGVR, ok := m.discovery.GetGVR("deployments")
	if !ok {
		return fmt.Errorf("deployments resource not found")
	}
	if _, err := m.dynClient.Resource(deployGVR).Namespace(namespace).Patch(ctx, name, types.JSONPatchType, patch, metav1.PatchOptions{}); err != nil {
		return rollbackPatchError("Deployment", namespace, name, err)
	}
	return nil
}

// rollbackControllerRevision applies a StatefulSet or DaemonSet ControllerRevision
// as a strategic merge patch, as kubectl rollout undo does. The controllers store
// data as {"spec":{"template":{..., "$patch":"replace"}}}, so the template is
// replaced rather than merged.
func (m *WorkloadManager) rollbackControllerRevision(ctx context.Context, normalizedKind string, workload *unstructured.Unstructured, revision int64) error {
	namespace, name := workload.GetNamespace(), workload.GetName()
	crGVR, ok := m.discovery.GetGVR("controllerrevisions")
	if !ok {
		return fmt.Errorf("controllerrevisions resource not found")
	}
	crList, err := m.dynClient.Resource(crGVR).Namespace(namespace).List(ctx, metav1.ListOptions{})
	if err != nil {
		return fmt.Errorf("failed to list controllerrevisions: %w", err)
	}

	var targetCR *unstructured.Unstructured
	for i := range crList.Items {
		cr := &crList.Items[i]
		if !ownedBy(cr, workload.GetUID()) {
			continue
		}
		revNum, found, err := unstructured.NestedInt64(cr.Object, "revision")
		if err == nil && found && revNum == revision {
			targetCR = cr
			break
		}
	}
	if targetCR == nil {
		return fmt.Errorf("revision %d not found", revision)
	}

	data, found, err := unstructured.NestedMap(targetCR.Object, "data")
	if err != nil || !found {
		return fmt.Errorf("failed to extract data from revision %d", revision)
	}

	if template, found, _ := unstructured.NestedMap(data, "spec", "template"); found {
		delete(template, "$patch")
		live, _, _ := unstructured.NestedMap(workload.Object, "spec", "template")
		if same, err := podTemplatesEqual(live, template, false); err != nil {
			return fmt.Errorf("failed to compare revision %d with the current template: %w", revision, err)
		} else if same {
			return fmt.Errorf("revision %d of %s %s/%s: %w", revision, workload.GetKind(), namespace, name, rollouts.ErrTemplateUnchanged)
		}
	}

	// A resourceVersion in the patch makes the API server reject it with a
	// conflict if the workload changed after it was read.
	data["metadata"] = map[string]any{"resourceVersion": workload.GetResourceVersion()}
	patchBytes, err := json.Marshal(data)
	if err != nil {
		return fmt.Errorf("failed to build rollback patch: %w", err)
	}

	gvr, ok := m.discovery.GetGVR(normalizedKind)
	if !ok {
		return fmt.Errorf("unknown resource kind: %s", normalizedKind)
	}
	if _, err := m.dynClient.Resource(gvr).Namespace(namespace).Patch(ctx, name, types.StrategicMergePatchType, patchBytes, metav1.PatchOptions{}); err != nil {
		return rollbackPatchError(workload.GetKind(), namespace, name, err)
	}
	return nil
}

func rollbackPatchError(kind, namespace, name string, err error) error {
	if apierrors.IsConflict(err) {
		return fmt.Errorf("%s %s/%s changed while rolling back; check its current template and retry: %w", kind, namespace, name, err)
	}
	return fmt.Errorf("failed to roll back %s %s/%s: %w", kind, namespace, name, err)
}

func ownedBy(obj *unstructured.Unstructured, uid types.UID) bool {
	for _, ref := range obj.GetOwnerReferences() {
		if ref.UID == uid {
			return true
		}
	}
	return false
}

// podTemplatesEqual compares templates as typed objects, the way the controllers
// and kubectl do, so equivalent quantities and absent-vs-empty fields are not
// mistaken for a difference. ignoreDeploymentHash drops pod-template-hash, which
// is Deployment controller bookkeeping and means nothing to other workloads.
func podTemplatesEqual(a, b map[string]any, ignoreDeploymentHash bool) (bool, error) {
	var ta, tb corev1.PodTemplateSpec
	if err := runtime.DefaultUnstructuredConverter.FromUnstructured(a, &ta); err != nil {
		return false, err
	}
	if err := runtime.DefaultUnstructuredConverter.FromUnstructured(b, &tb); err != nil {
		return false, err
	}
	if ignoreDeploymentHash {
		delete(ta.Labels, appsv1.DefaultDeploymentUniqueLabelKey)
		delete(tb.Labels, appsv1.DefaultDeploymentUniqueLabelKey)
	}
	return apiequality.Semantic.DeepEqual(ta, tb), nil
}

// NormalizeWorkloadKind converts various kind formats to the plural lowercase form.
func NormalizeWorkloadKind(kind string) string {
	switch kind {
	case "Deployment", "deployment", "deployments":
		return "deployments"
	case "StatefulSet", "statefulset", "statefulsets":
		return "statefulsets"
	case "DaemonSet", "daemonset", "daemonsets":
		return "daemonsets"
	case "Rollout", "rollout", "rollouts":
		return "rollouts"
	default:
		return kind
	}
}

func extractContainerImage(obj map[string]any) string {
	containers, found, _ := unstructured.NestedSlice(obj, "spec", "template", "spec", "containers")
	if !found || len(containers) == 0 {
		return ""
	}
	if container, ok := containers[0].(map[string]any); ok {
		if image, ok := container["image"].(string); ok {
			return image
		}
	}
	return ""
}

func extractContainerImageFromData(obj map[string]any) string {
	data, found, _ := unstructured.NestedMap(obj, "data")
	if !found {
		return ""
	}
	return extractContainerImage(data)
}