package k8s

import (
	"bufio"
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"strconv"
	"strings"
	"time"

	brv1alpha1 "github.com/pingcap/tidb-operator/api/v2/br/v1alpha1"
	"github.com/pingcap/tidb-operator/api/v2/core/v1alpha1"
	schemav1 "github.com/tidbcloud/cluster-service-ng/pkg/storage/etcd/schema/v1"
	"gomodules.xyz/jsonpatch/v3"
	appsv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	apimeta "k8s.io/apimachinery/pkg/api/meta"
	"k8s.io/apimachinery/pkg/api/resource"
	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"
	utilyaml "k8s.io/apimachinery/pkg/util/yaml"
	"k8s.io/client-go/rest"
	"sigs.k8s.io/controller-runtime/pkg/cache"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/yaml"
)

const (
	SharedPoolVCPUCeilingResourceQuotaName    = "sharedpool-vcpu-ceiling"
	SharedPoolDefaultCPURequestLimitRangeName = "sharedpool-default-cpu-request"
	sharedPoolDefaultCPURequest               = "20m"
)

func k8sClientCacheDisabledObjects() []client.Object {
	return []client.Object{
		&schemav1.Task{},
		&v1alpha1.Cluster{},
		&v1alpha1.TiDBGroup{},
		&v1alpha1.TiKVGroup{},
		&v1alpha1.PlacementPolicy{},
		&v1alpha1.TiFlashGroup{},
		&v1alpha1.PDGroup{},
		&v1alpha1.TSOGroup{},
		&v1alpha1.SchedulingGroup{},
		&v1alpha1.TiCDCGroup{},
		&v1alpha1.TiProxyGroup{},
		&v1alpha1.TiKVWorkerGroup{},
		&v1alpha1.ResourceManagerGroup{},
		&brv1alpha1.TiBR{},
		&brv1alpha1.TiBRGC{},
		&corev1.Service{},
		&corev1.Secret{},
		&appsv1.Deployment{},
		&corev1.ConfigMap{},
		&corev1.ResourceQuota{},
		&corev1.LimitRange{},
	}
}

func NewClient(ctx context.Context, cfg *rest.Config) (client.Client, error) {
	// not set k8s client timeout
	cfg.Timeout = 0
	cfg.QPS = 100
	cfg.Burst = 150

	hc, err := rest.HTTPClientFor(cfg)
	if err != nil {
		return nil, err
	}

	reader, err := cache.New(cfg, cache.Options{
		Scheme:     scheme,
		HTTPClient: hc,
	})
	if err != nil {
		return nil, err
	}

	//nolint:errcheck // it will not return any errors
	go reader.Start(ctx)

	nctx, cancel := context.WithTimeout(ctx, time.Second*30)
	defer cancel()
	if !reader.WaitForCacheSync(nctx) {
		return nil, fmt.Errorf("cache is not synced: %w", nctx.Err())
	}

	c, err := client.New(cfg, client.Options{
		Scheme:     scheme,
		HTTPClient: hc,
		Cache: &client.CacheOptions{
			Reader:       reader,
			DisableFor:   k8sClientCacheDisabledObjects(),
			Unstructured: false,
		},
	})
	if err != nil {
		return nil, err
	}

	return c, nil
}

func DeleteObject(ctx context.Context, cli client.Writer, obj client.Object) error {
	if err := client.IgnoreNotFound(cli.Delete(ctx, obj)); err != nil {
		return err
	}
	return nil
}

func CreateIfNotExist(ctx context.Context, cli client.Client, obj client.Object) error {
	err := cli.Get(ctx, client.ObjectKeyFromObject(obj), obj)
	if err != nil {
		if apierrors.IsNotFound(err) {
			return cli.Create(ctx, obj)
		}
		return err
	}

	return nil
}

// ErrMultiDocYAML is returned by GenerateUnstructuredFromYaml when the input
// contains more than one non-empty YAML document. csng's apply path renders
// one .templ per kind and feeds it through this function, so a multi-doc
// input always indicates a template authoring mistake; failing fast is much
// better than the previous behaviour of sigs.k8s.io/yaml.YAMLToJSON, which
// silently kept only the first document and dropped the rest. See #2553 for
// the cassandra-proxy regression that motivated this guard.
var ErrMultiDocYAML = errors.New("multi-document YAML is not supported: each .templ must hold exactly one Kind")

func GenerateUnstructuredFromYaml(yamlStr string) (*unstructured.Unstructured, *schema.GroupVersionKind, error) {
	if err := assertSingleYAMLDoc(yamlStr); err != nil {
		return nil, nil, err
	}

	tcJson, err := yaml.YAMLToJSON([]byte(yamlStr))
	if err != nil {
		return nil, nil, fmt.Errorf("failed to convert yaml to json: %w", err)
	}

	obj, gvk, err := unstructured.UnstructuredJSONScheme.Decode(tcJson, nil, nil)
	if err != nil {
		return nil, nil, fmt.Errorf("failed to decode json to unstructured: %w", err)
	}

	unstructuredObj, err := runtime.DefaultUnstructuredConverter.ToUnstructured(obj)
	if err != nil {
		return nil, nil, fmt.Errorf("failed to convert to unstructured: %w", err)
	}

	var unstruct unstructured.Unstructured

	unstruct.Object = unstructuredObj

	return &unstruct, gvk, nil
}

// assertSingleYAMLDoc returns ErrMultiDocYAML when yamlStr contains more than
// one non-empty YAML document. Empty documents (a stray leading or trailing
// `---` separator with nothing meaningful between separators) are tolerated
// because YAML emitters add them harmlessly.
func assertSingleYAMLDoc(yamlStr string) error {
	reader := utilyaml.NewYAMLReader(bufio.NewReader(bytes.NewReader([]byte(yamlStr))))
	seen := 0
	for {
		doc, err := reader.Read()
		if errors.Is(err, io.EOF) {
			return nil
		}
		if err != nil {
			return fmt.Errorf("failed to scan yaml documents: %w", err)
		}
		// k8s NewYAMLReader keeps the leading `---` separator on the chunk it
		// returns. Strip it (plus surrounding whitespace) before deciding
		// whether the chunk carries any payload, otherwise a chunk that is
		// just `---\n` would be miscounted as a real document.
		body := bytes.TrimSpace(doc)
		body = bytes.TrimPrefix(body, []byte("---"))
		body = bytes.TrimSpace(body)
		if len(body) == 0 {
			continue
		}
		seen++
		if seen > 1 {
			return ErrMultiDocYAML
		}
	}
}

func EnsureNamespace(ctx context.Context, k8sClient client.Client, namespaceName, tenantId string) error {
	namespace := &corev1.Namespace{}
	err := k8sClient.Get(ctx, client.ObjectKey{Name: namespaceName}, namespace)

	labels := map[string]string{
		schemav1.LabelUsedForKey: schemav1.ClusterNGServiceType,
	}

	if tenantId != "" {
		labels[schemav1.LabelUserTenantKey] = tenantId
	}

	if err != nil {
		if apierrors.IsNotFound(err) {
			return k8sClient.Create(ctx, &corev1.Namespace{
				ObjectMeta: metav1.ObjectMeta{
					Name:   namespaceName,
					Labels: labels,
				},
			})
		}
		return err
	}

	if namespace.Status.Phase == corev1.NamespaceTerminating {
		return fmt.Errorf("namespace %s is terminating, wait for it to be deleted and try again", namespaceName)
	}

	return nil
}

func EnsureResourceQuota(ctx context.Context, k8sClient client.Client, namespaceName string, vcpusLimit uint64) error {
	resourceQuota := &corev1.ResourceQuota{
		ObjectMeta: metav1.ObjectMeta{
			Name:      SharedPoolVCPUCeilingResourceQuotaName,
			Namespace: namespaceName,
		},
	}

	if vcpusLimit == 0 {
		return client.IgnoreNotFound(k8sClient.Delete(ctx, resourceQuota))
	}

	_, err := controllerutil.CreateOrUpdate(ctx, k8sClient, resourceQuota, func() error {
		resourceQuota.Spec.Hard = corev1.ResourceList{
			corev1.ResourceRequestsCPU: *resource.NewQuantity(int64(vcpusLimit), resource.DecimalSI),
		}
		return nil
	})
	if err != nil {
		return err
	}

	return nil
}

func EnsureSharedPoolLimitRange(ctx context.Context, k8sClient client.Client, namespaceName string, vcpusLimit uint64) error {
	limitRange := &corev1.LimitRange{
		ObjectMeta: metav1.ObjectMeta{
			Name:      SharedPoolDefaultCPURequestLimitRangeName,
			Namespace: namespaceName,
		},
	}

	if vcpusLimit == 0 {
		return client.IgnoreNotFound(k8sClient.Delete(ctx, limitRange))
	}

	defaultCPURequest := resource.MustParse(sharedPoolDefaultCPURequest)
	_, err := controllerutil.CreateOrUpdate(ctx, k8sClient, limitRange, func() error {
		limitRange.Spec.Limits = []corev1.LimitRangeItem{{
			Type: corev1.LimitTypeContainer,
			DefaultRequest: corev1.ResourceList{
				corev1.ResourceCPU: defaultCPURequest,
			},
		}}
		return nil
	})
	if err != nil {
		return err
	}

	return nil
}

func RESTMapper() apimeta.RESTMapper {
	gvs := scheme.PreferredVersionAllGroups()
	mapper := apimeta.NewDefaultRESTMapper(gvs)
	// Optionally add known types
	// there are multiple Deployment in the scheme, we only care Deployment and DeploymentList with apps/v1
	for gvk := range scheme.AllKnownTypes() {
		apiversion, kind := gvk.ToAPIVersionAndKind()
		if kind == "Deployment" && apiversion != "apps/v1" {
			continue
		}
		if kind == "DeploymentList" && apiversion != "apps/v1" {
			continue
		}
		// there are multiple HorizontalPodAutoscaler in the scheme, we only care HorizontalPodAutoscaler and HorizontalPodAutoscalerList with autoscaling/v2
		if kind == "HorizontalPodAutoscaler" && apiversion != "autoscaling/v2" {
			continue
		}
		if kind == "HorizontalPodAutoscalerList" && apiversion != "autoscaling/v2" {
			continue
		}
		mapper.Add(gvk, apimeta.RESTScopeNamespace)
	}
	return mapper
}

// CompareK8sVersion compares two k8s version strings.
// Returns: 1 if v1 > v2, -1 if v1 < v2, 0 if v1 == v2
// Supports formats like "1.34", "1.34.1", "1.34.1-aliyun.1"
// Returns error if version parts cannot be parsed as integers.
func CompareK8sVersion(v1, v2 string) (int, error) {
	// normalize: remove prefix like "v" and suffix like "-aliyun.1"
	normalize := func(v string) string {
		v = strings.TrimPrefix(v, "v")
		if idx := strings.IndexAny(v, "-+"); idx != -1 {
			v = v[:idx]
		}
		return v
	}

	parts1 := strings.Split(normalize(v1), ".")
	parts2 := strings.Split(normalize(v2), ".")

	// compare each part
	maxLen := len(parts1)
	if len(parts2) > maxLen {
		maxLen = len(parts2)
	}

	for i := 0; i < maxLen; i++ {
		var n1, n2 int
		var err error
		if i < len(parts1) {
			n1, err = strconv.Atoi(parts1[i])
			if err != nil {
				return 0, fmt.Errorf("failed to parse version part %q from version %q: %w", parts1[i], v1, err)
			}
		}
		if i < len(parts2) {
			n2, err = strconv.Atoi(parts2[i])
			if err != nil {
				return 0, fmt.Errorf("failed to parse version part %q from version %q: %w", parts2[i], v2, err)
			}
		}

		if n1 > n2 {
			return 1, nil
		}
		if n1 < n2 {
			return -1, nil
		}
	}

	return 0, nil
}

func GetPodCondition(status *corev1.PodStatus, conditionType corev1.PodConditionType) (int, *corev1.PodCondition) {
	if status == nil {
		return -1, nil
	}
	return GetPodConditionFromList(status.Conditions, conditionType)
}

func GetPodConditionFromList(conditions []corev1.PodCondition, conditionType corev1.PodConditionType) (int, *corev1.PodCondition) {
	if conditions == nil {
		return -1, nil
	}
	for i := range conditions {
		if conditions[i].Type == conditionType {
			return i, &conditions[i]
		}
	}
	return -1, nil
}

func ConstructPatch(patchPtrs ...*jsonpatch.Operation) (client.Patch, error) {
	patches := make([]jsonpatch.Operation, 0, len(patchPtrs))
	for _, patchPtr := range patchPtrs {
		if patchPtr != nil {
			patches = append(patches, *patchPtr)
		}
	}
	buf, err := json.Marshal(patches)
	if err != nil {
		return nil, err
	}
	return client.RawPatch(types.JSONPatchType, buf), nil
}