package cluster

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"time"

	"github.com/pingcap/log"
	"go.uber.org/zap"
	corev1 "k8s.io/api/core/v1"
	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/fields"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/rest"
	"sigs.k8s.io/controller-runtime/pkg/cache"
	"sigs.k8s.io/controller-runtime/pkg/client"

	"github.com/tidbcloud/clusterproxy/resolve"
	"github.com/tidbcloud/clusterproxy/route"
	"github.com/tidbcloud/clusterproxy/route/router"
)

const (
	operateTypeCreate       = "Create"
	operateTypeGet          = "Get"
	operateTypeGetDirectly  = "GetDirectly"
	operateTypeList         = "List"
	operateTypeDelete       = "Delete"
	operateTypeUpdate       = "Update"
	operateTypeUpdateStatus = "UpdateStatus"

	branchAPIVersion = "cloud.pingcap.com/v1"
	branchKind       = "Branch"
	branchPhaseField = "status.phase"
)

var (
	directClient client.Client
	cacheClient  client.Client
)

// CreateK8sResourceClients initializes the singleton clients at startup.
// It is safe to call multiple times.
func CreateK8sResourceClients(cfg *rest.Config) error {
	if cfg == nil {
		var err error
		cfg, err = rest.InClusterConfig()
		if err != nil {
			return err
		}
	} else {
		cfg = rest.CopyConfig(cfg)
	}

	var err error
	directClient, err = client.New(rest.CopyConfig(cfg), client.Options{})
	if err != nil {
		return err
	}
	cacheClient, err = newCachedClient(rest.CopyConfig(cfg))
	if err != nil {
		return err
	}

	return nil
}

func init() {
	router.RegisterRoute(k8sResourceRoute)
}

var k8sResourceRoute = &route.Route{
	Name:    "k8s-resource",
	Path:    "k8s-resource",
	Version: route.V1,
	Schema: &route.Schema{
		PathRegex:       "^$",
		RequiredHeaders: []string{},
		JSONBodyModel:   &k8sResourceModel{},
	},
	GetHandler: k8sResourceGetHandler,
}

// k8sResourceModel is the request body for the k8s-resource route.
//
// Required fields by operateType:
//   - Create: operateType, apiVersion, kind, namespace, object
//   - Get/GetDirectly/Delete: operateType, apiVersion, kind, namespace, name
//   - List: operateType, apiVersion, kind, namespace
//   - Update: operateType, apiVersion, kind, namespace, object
//   - UpdateStatus: operateType, apiVersion, kind, namespace, name, status
//
// Notes:
//   - name is optional for Create/Update, and is only used as a fallback when
//     object.metadata.name is omitted.
//   - labels and fieldSelector are only used by List.
type k8sResourceModel struct {
	// Required: used by all operations to select Create, Get, GetDirectly,
	// List, Delete, Update, or UpdateStatus.
	OperateType string `json:"operateType"`
	// Required: used by all operations to identify the target resource version.
	APIVersion string `json:"apiVersion"`
	// Required: used by all operations to identify the target resource kind.
	Kind string `json:"kind"`
	// Required: used by all operations to scope the target resource.
	Namespace string `json:"namespace"`
	// Required by Get, GetDirectly, Delete, and UpdateStatus.
	// Optional for Create and Update: only used when object.metadata.name is
	// empty.
	Name string `json:"name"`
	// Optional: only used to filter List requests.
	Labels map[string]string `json:"labels"`
	// Optional: only used to filter List requests. It currently only supports
	// status.phase on cloud.pingcap.com/v1 Branch resources.
	FieldSelector map[string]string `json:"fieldSelector"`
	// Required by Create and Update: the full resource body.
	Object map[string]interface{} `json:"object"`
	// Required by UpdateStatus: the status subresource payload.
	Status map[string]interface{} `json:"status"`
}

// k8sResourceResponse is the JSON envelope returned by the k8s-resource route.
//
// Clients should decode successful responses as:
//   - {"status": <http-status>} for Create, Delete, Update, and UpdateStatus
//   - {"status": <http-status>, "data": <object>} for Get and GetDirectly
//   - {"status": <http-status>, "data": []object} for List
//
// Error responses do not use this envelope. Clients should treat any non-2xx
// response as plain text returned by http.Error, where the body is err.Error()
// plus a trailing newline.
type k8sResourceResponse struct {
	Status int         `json:"status"`
	Data   interface{} `json:"data,omitempty"`
}

func k8sResourceGetHandler(_ *route.Context, cl resolve.Cluster) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		body, err := io.ReadAll(r.Body)
		defer r.Body.Close() // nolint
		if err != nil {
			http.Error(w, "failed to read body contents", http.StatusInternalServerError)
			return
		}

		req := &k8sResourceModel{}
		if err := json.Unmarshal(body, req); err != nil {
			http.Error(w, fmt.Sprintf("failed to parse body contents: %v", err), http.StatusBadRequest)
			return
		}

		if err := validateK8sResourceRequest(req); err != nil {
			http.Error(w, err.Error(), http.StatusBadRequest)
			return
		}

		reqLogger := log.L().With(
			zap.String("operate_type", req.OperateType),
			zap.String("api_version", req.APIVersion),
			zap.String("kind", req.Kind),
			zap.String("namespace", req.Namespace),
			zap.String("name", req.Name),
		)

		if directClient == nil {
			http.Error(w, "in-cluster controller-runtime client is not initialized", http.StatusInternalServerError)
			return
		}
		if cacheClient == nil {
			http.Error(w, "in-cluster cache client is not initialized", http.StatusInternalServerError)
			return
		}

		ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
		defer cancel()

		resp, statusCode, err := executeK8sResourceOperation(ctx, cacheClient, directClient, req)
		if err != nil {
			reqLogger.Error("k8s-resource operation failed",
				zap.Int("status_code", statusCode),
				zap.Error(err),
			)
			http.Error(w, err.Error(), statusCode)
			return
		}
		if committed, err := writeK8sResourceResponse(w, resp, statusCode); err != nil {
			reqLogger.Error("k8s-resource write response failed",
				zap.Int("status_code", statusCode),
				zap.Error(err),
			)
			if !committed {
				http.Error(w, err.Error(), http.StatusInternalServerError)
			}
		}
	})
}

func newCachedClient(cfg *rest.Config) (client.Client, error) {
	c, err := cache.New(cfg, cache.Options{})
	if err != nil {
		return nil, err
	}
	branchObj := &unstructured.Unstructured{}
	branchObj.SetAPIVersion(branchAPIVersion)
	branchObj.SetKind(branchKind)
	if err = c.IndexField(context.Background(), branchObj, branchPhaseField, func(obj client.Object) []string {
		un, ok := obj.(*unstructured.Unstructured)
		if !ok {
			return nil
		}
		phase, found, nestedErr := unstructured.NestedString(un.Object, "status", "phase")
		if nestedErr != nil || !found || phase == "" {
			return nil
		}
		return []string{phase}
	}); err != nil {
		return nil, fmt.Errorf("index field %s for branch failed: %w", branchPhaseField, err)
	}

	// cache.Start will block until context is canceled.
	go func() {
		_ = c.Start(context.Background())
	}()

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

	return client.New(cfg, client.Options{
		Cache: &client.CacheOptions{
			Reader: c,
		},
	})
}

func validateK8sResourceRequest(req *k8sResourceModel) error {
	if req.OperateType == "" {
		return fmt.Errorf("operateType is required")
	}
	if req.APIVersion == "" || req.Kind == "" {
		return fmt.Errorf("apiVersion and kind are required")
	}

	// allow list for all namespace
	if req.Namespace == "" && req.OperateType != operateTypeList {
		return fmt.Errorf("namespace is required")
	}

	switch req.OperateType {
	case operateTypeCreate:
		if len(req.Object) == 0 {
			return fmt.Errorf("object is required for Create")
		}
	case operateTypeGet, operateTypeGetDirectly, operateTypeDelete:
		if req.Name == "" {
			return fmt.Errorf("name is required for %s", req.OperateType)
		}
	case operateTypeList:
		// List does not require name.
	case operateTypeUpdate:
		if len(req.Object) == 0 {
			return fmt.Errorf("object is required for Update")
		}
	case operateTypeUpdateStatus:
		if req.Name == "" {
			return fmt.Errorf("name is required for UpdateStatus")
		}
		if len(req.Status) == 0 {
			return fmt.Errorf("status is required for UpdateStatus")
		}
	default:
		return fmt.Errorf("unknown operateType %s", req.OperateType)
	}
	return nil
}

func executeK8sResourceOperation(
	ctx context.Context,
	cacheCli client.Reader,
	directCli client.Client,
	req *k8sResourceModel,
) (interface{}, int, error) {
	gvk, err := parseGVK(req.APIVersion, req.Kind)
	if err != nil {
		return nil, http.StatusBadRequest, err
	}

	switch req.OperateType {
	case operateTypeCreate:
		obj, err := newObjectForWrite(req, gvk)
		if err != nil {
			return nil, http.StatusBadRequest, err
		}
		if err = ensureNamespaceExists(ctx, directCli, req.Namespace); err != nil {
			return nil, mapErrorStatusCode(err), err
		}
		if err = directCli.Create(ctx, obj); err != nil {
			if !apierrors.IsAlreadyExists(err) {
				return nil, mapErrorStatusCode(err), err
			}
		}
		return nil, http.StatusOK, nil
	case operateTypeGet:
		obj := &unstructured.Unstructured{}
		obj.SetGroupVersionKind(gvk)
		// operateTypeGet is a cache-only read. No direct fallback.
		err = cacheCli.Get(ctx, types.NamespacedName{Namespace: req.Namespace, Name: req.Name}, obj)
		if err != nil {
			return nil, mapErrorStatusCode(err), err
		}
		return obj.Object, http.StatusOK, nil
	case operateTypeGetDirectly:
		obj := &unstructured.Unstructured{}
		obj.SetGroupVersionKind(gvk)
		err = directCli.Get(ctx, types.NamespacedName{Namespace: req.Namespace, Name: req.Name}, obj)
		if err != nil {
			return nil, mapErrorStatusCode(err), err
		}
		return obj.Object, http.StatusOK, nil
	case operateTypeList:
		list := &unstructured.UnstructuredList{}
		list.SetGroupVersionKind(gvk.GroupVersion().WithKind(gvk.Kind + "List"))
		listOpts := []client.ListOption{
			client.InNamespace(req.Namespace),
		}
		if len(req.Labels) > 0 {
			listOpts = append(listOpts, client.MatchingLabels(req.Labels))
		}
		if len(req.FieldSelector) > 0 {
			if err = validateSupportedFieldSelector(req); err != nil {
				return nil, http.StatusBadRequest, err
			}
			selector := fields.SelectorFromSet(req.FieldSelector)
			listOpts = append(listOpts, client.MatchingFieldsSelector{Selector: selector})
		}
		err = cacheCli.List(ctx, list, listOpts...)
		if err != nil {
			return nil, mapErrorStatusCode(err), err
		}
		items := make([]interface{}, 0, len(list.Items))
		for i := range list.Items {
			items = append(items, list.Items[i].Object)
		}
		return items, http.StatusOK, nil
	case operateTypeDelete:
		obj := &unstructured.Unstructured{}
		obj.SetGroupVersionKind(gvk)
		obj.SetNamespace(req.Namespace)
		obj.SetName(req.Name)
		if err = directCli.Delete(ctx, obj, client.PropagationPolicy(metav1.DeletePropagationBackground)); err != nil {
			if !apierrors.IsNotFound(err) {
				return nil, mapErrorStatusCode(err), err
			}
		}
		return nil, http.StatusOK, nil
	case operateTypeUpdate:
		obj, err := newObjectForWrite(req, gvk)
		if err != nil {
			return nil, http.StatusBadRequest, err
		}
		if err = directCli.Update(ctx, obj); err != nil {
			return nil, mapErrorStatusCode(err), err
		}
		return nil, http.StatusOK, nil
	case operateTypeUpdateStatus:
		obj := &unstructured.Unstructured{}
		obj.SetGroupVersionKind(gvk)
		err = directCli.Get(ctx, types.NamespacedName{Namespace: req.Namespace, Name: req.Name}, obj)
		if err != nil {
			return nil, mapErrorStatusCode(err), err
		}
		obj.Object["status"] = req.Status
		if err = directCli.Status().Update(ctx, obj); err != nil {
			return nil, mapErrorStatusCode(err), err
		}
		return nil, http.StatusOK, nil
	default:
		return nil, http.StatusBadRequest, fmt.Errorf("unknown operateType %s", req.OperateType)
	}
}

func validateSupportedFieldSelector(req *k8sResourceModel) error {
	if req.APIVersion != branchAPIVersion || req.Kind != branchKind {
		return fmt.Errorf("fieldSelector currently only supports %s for %s %s", branchPhaseField, branchAPIVersion, branchKind)
	}
	for key, value := range req.FieldSelector {
		if key != branchPhaseField {
			return fmt.Errorf("fieldSelector currently only supports %s conditions, got %s", branchPhaseField, key)
		}
		if value == "" {
			return fmt.Errorf("fieldSelector value for %s must not be empty", key)
		}
	}
	return nil
}

func parseGVK(apiVersion, kind string) (schema.GroupVersionKind, error) {
	gv, err := schema.ParseGroupVersion(apiVersion)
	if err != nil {
		return schema.GroupVersionKind{}, fmt.Errorf("invalid apiVersion %s: %w", apiVersion, err)
	}
	if kind == "" {
		return schema.GroupVersionKind{}, fmt.Errorf("kind is required")
	}
	return gv.WithKind(kind), nil
}

func newObjectForWrite(req *k8sResourceModel, gvk schema.GroupVersionKind) (*unstructured.Unstructured, error) {
	if len(req.Object) == 0 {
		return nil, fmt.Errorf("object is required")
	}
	obj := &unstructured.Unstructured{Object: req.Object}
	obj.SetGroupVersionKind(gvk)
	obj.SetNamespace(req.Namespace)
	if req.Name != "" && obj.GetName() == "" {
		obj.SetName(req.Name)
	}
	return obj, nil
}

func ensureNamespaceExists(ctx context.Context, directCli client.Client, namespace string) error {
	// The namespace is created idempotently to avoid "namespace not found" on create.
	ns := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: namespace}}
	if err := directCli.Create(ctx, ns); err != nil {
		if !apierrors.IsAlreadyExists(err) {
			log.Error("ensure namespace exists failed", zap.String("namespace", namespace), zap.Error(err))
			return err
		}
	}
	return nil
}

func mapErrorStatusCode(err error) int {
	switch {
	case apierrors.IsNotFound(err):
		return http.StatusNotFound
	case apierrors.IsForbidden(err):
		return http.StatusForbidden
	case apierrors.IsConflict(err):
		return http.StatusConflict
	case apierrors.IsUnauthorized(err):
		return http.StatusUnauthorized
	case apierrors.IsBadRequest(err):
		return http.StatusBadRequest
	default:
		return http.StatusInternalServerError
	}
}

func writeK8sResourceResponse(w http.ResponseWriter, data interface{}, statusCode int) (bool, error) {
	// Some old call paths may pass a zero status. Normalize to 200 for compatibility.
	if statusCode == 0 {
		statusCode = http.StatusOK
	}
	resp, err := json.Marshal(&k8sResourceResponse{Status: statusCode, Data: data})
	if err != nil {
		return false, err
	}
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(statusCode)
	_, err = w.Write(resp)
	return true, err
}