// Package executor applies autoscaling decisions to Kubernetes targets.
package executor

import (
	"context"
	"encoding/json"
	"fmt"

	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/util/retry"
	"sigs.k8s.io/controller-runtime/pkg/client"

	"github.com/tidbcloud/cluster-service-ng/internal/capacity-planner/pkg/merger"
	"github.com/tidbcloud/cluster-service-ng/internal/capacity-planner/pkg/tier"
)

// Target identifies the Kubernetes object that receives autoscaling patches.
type Target struct {
	Namespace  string
	APIVersion string
	Kind       string
	Name       string
}

// Executor updates a target object to the supplied tier and replica count.
type Executor interface {
	Execute(ctx context.Context, target *Target, targetTier tier.Tier, replicas int32) error
}

// KubernetesExecutor merges desired target capacity into the latest target and
// submits one optimistic-lock merge patch.
type KubernetesExecutor struct {
	reader client.Reader
	writer client.Client
	merger merger.Interface
}

// NewKubernetesExecutor creates an executor with separate uncached read and
// write clients. objectMerger preserves fields outside the scaling decision
// according to the live target CRD schema.
func NewKubernetesExecutor(reader client.Reader, writer client.Client, objectMerger merger.Interface) *KubernetesExecutor {
	return &KubernetesExecutor{reader: reader, writer: writer, merger: objectMerger}
}

var _ Executor = (*KubernetesExecutor)(nil)

// Execute merges the target tier and spec.replicas into the latest target.
func (e *KubernetesExecutor) Execute(ctx context.Context, target *Target, targetTier tier.Tier, replicas int32) error {
	return e.execute(ctx, target, targetTier, replicas)
}

// execute uses a read-modify-patch loop because CRDs can combine atomic maps
// and list-map fields in one tier patch. No single server-side patch strategy
// preserves both without first merging against the live object.
func (e *KubernetesExecutor) execute(ctx context.Context, target *Target, targetTier tier.Tier, replicas int32) error {
	if e == nil || e.reader == nil || e.writer == nil {
		return fmt.Errorf("executor: nil Kubernetes reader or writer")
	}
	if e.merger == nil {
		return fmt.Errorf("executor: nil object merger")
	}
	if target == nil {
		return fmt.Errorf("executor: nil target")
	}
	if target.Namespace == "" || target.APIVersion == "" || target.Kind == "" || target.Name == "" {
		return fmt.Errorf("executor: target namespace, apiVersion, kind and name are required")
	}
	if targetTier == nil {
		return fmt.Errorf("executor: nil tier")
	}

	patch := targetTier.Patch()
	if patch == nil {
		return fmt.Errorf("executor: nil patch for tier %q", targetTier.Name())
	}
	// Tier templates are rendered as merge-patch-shaped JSON. The merger treats
	// the document as partial desired state rather than sending it directly.
	if patch.Type() != types.MergePatchType {
		return fmt.Errorf("executor: tier %q patch type %q is unsupported", targetTier.Name(), patch.Type())
	}
	gv, err := schema.ParseGroupVersion(target.APIVersion)
	if err != nil {
		return fmt.Errorf("executor: target apiVersion %q: %w", target.APIVersion, err)
	}
	gvk := gv.WithKind(target.Kind)
	desired, err := buildDesiredObject(patch, gvk, replicas)
	if err != nil {
		return fmt.Errorf("executor: tier %q patch: %w", targetTier.Name(), err)
	}

	err = retry.RetryOnConflict(retry.DefaultRetry, func() error {
		live := &unstructured.Unstructured{}
		live.SetGroupVersionKind(gvk)
		key := client.ObjectKey{Namespace: target.Namespace, Name: target.Name}
		if err := e.reader.Get(ctx, key, live); err != nil {
			return err
		}
		base := live.DeepCopy()
		if err := e.merger.Merge(ctx, gvk, live.Object, desired); err != nil {
			return err
		}
		return e.writer.Patch(ctx, live, client.MergeFromWithOptions(
			base,
			client.MergeFromWithOptimisticLock{},
		))
	})
	if err != nil {
		return fmt.Errorf("merge tier %q into %s/%s %s/%s: %w",
			targetTier.Name(), target.APIVersion, target.Kind, target.Namespace, target.Name, err)
	}
	return nil
}

func buildDesiredObject(
	patch client.Patch,
	gvk schema.GroupVersionKind,
	replicas int32,
) (map[string]interface{}, error) {
	obj := &unstructured.Unstructured{}
	obj.SetGroupVersionKind(gvk)

	data, err := patch.Data(obj)
	if err != nil {
		return nil, fmt.Errorf("build tier patch: %w", err)
	}
	desired := map[string]interface{}{}
	if len(data) > 0 {
		if err := json.Unmarshal(data, &desired); err != nil {
			return nil, fmt.Errorf("decode tier patch: %w", err)
		}
	}
	mergeMap(desired, map[string]interface{}{
		"spec": map[string]interface{}{
			"replicas": int64(replicas),
		},
	})

	// Identity and server-populated metadata belong to the live object, never
	// to a reusable tier template.
	delete(desired, "apiVersion")
	delete(desired, "kind")
	if metadata, ok := desired["metadata"].(map[string]interface{}); ok {
		for _, key := range []string{
			"name", "namespace", "uid", "resourceVersion", "generation",
			"creationTimestamp", "deletionTimestamp", "managedFields",
		} {
			delete(metadata, key)
		}
	}
	return desired, nil
}

func mergeMap(dst, src map[string]interface{}) {
	for key, srcValue := range src {
		srcMap, srcIsMap := srcValue.(map[string]interface{})
		dstMap, dstIsMap := dst[key].(map[string]interface{})
		if srcIsMap && dstIsMap {
			mergeMap(dstMap, srcMap)
			continue
		}
		dst[key] = srcValue
	}
}