package k8sresource

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"strings"

	"github.com/pingcap/errors"
	regionalpb "github.com/tidbcloud/infra-provider/api/tidbcloud/regional-server/v1"
	bizerr "github.com/tidbcloud/infra-provider/regional-server/pkg/common/errors"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/meta"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/client-go/kubernetes/scheme"
	"k8s.io/client-go/rest"
	"sigs.k8s.io/controller-runtime/pkg/client/apiutil"
)

// Table-only Accept: no application/json fallback so unsupported types fail instead of returning full objects.
const tableAcceptHeader = "application/json;as=Table;v=v1;g=meta.k8s.io"

func getAsTable(
	ctx context.Context,
	restConfig *rest.Config,
	mapper meta.RESTMapper,
	gvk schema.GroupVersionKind,
	namespace, name string,
) (*metav1.Table, error) {
	if restConfig == nil {
		return nil, bizerr.New(bizerr.K8SClientError, "REST config is not available")
	}
	mapping, err := mapper.RESTMapping(gvk.GroupKind(), gvk.Version)
	if err != nil {
		return nil, mapRESTMapperError(err, gvk.Group, gvk.Kind)
	}
	return requestTable(ctx, restConfig, mapping, gvk, namespace, name, nil)
}

func listAsTable(
	ctx context.Context,
	restConfig *rest.Config,
	mapper meta.RESTMapper,
	gvk schema.GroupVersionKind,
	sel *regionalpb.Selector,
	limit int64,
) (*metav1.Table, error) {
	if restConfig == nil {
		return nil, bizerr.New(bizerr.K8SClientError, "REST config is not available")
	}
	mapping, err := mapper.RESTMapping(gvk.GroupKind(), gvk.Version)
	if err != nil {
		return nil, mapRESTMapperError(err, gvk.Group, gvk.Kind)
	}
	namespace := ""
	if sel != nil {
		namespace = strings.TrimSpace(sel.Namespace)
	}
	listOpts := &metav1.ListOptions{Limit: limit}
	if labelSelector := labelSelectorString(sel); labelSelector != "" {
		listOpts.LabelSelector = labelSelector
	}
	return requestTable(ctx, restConfig, mapping, gvk, namespace, "", listOpts)
}

func requestTable(
	ctx context.Context,
	restConfig *rest.Config,
	mapping *meta.RESTMapping,
	gvk schema.GroupVersionKind,
	namespace, name string,
	listOpts *metav1.ListOptions,
) (*metav1.Table, error) {
	restClient, err := newResourceRESTClient(restConfig, gvk)
	if err != nil {
		return nil, bizerr.New(bizerr.K8SClientError, errors.Wrap(err, "create REST client").Error())
	}

	req := restClient.Get().
		SetHeader("Accept", tableAcceptHeader).
		Resource(mapping.Resource.Resource)
	if mapping.Scope.Name() == meta.RESTScopeNameNamespace {
		req = req.Namespace(namespace)
	}
	if name != "" {
		req = req.Name(name)
	} else if listOpts != nil {
		// ListOptions belong to meta API; encode against meta.k8s.io/v1, not the resource GV.
		// VersionedParams + scheme.ParameterCodec fails for CRDs (e.g. APPBox) whose GV is
		// not registered in kubernetes scheme.
		req = req.SpecificallyVersionedParams(listOpts, metav1.ParameterCodec, metav1.SchemeGroupVersion)
	}

	body, err := req.DoRaw(ctx)
	if err != nil {
		return nil, mapTableRequestError(err)
	}
	table := &metav1.Table{}
	if err := json.Unmarshal(body, table); err != nil {
		return nil, tableConversionUnavailable(err)
	}
	if table.Kind != "Table" {
		return nil, tableConversionUnavailable(fmt.Errorf("response kind is %q, expected Table", table.Kind))
	}
	if err := decodeTableRowObjects(table); err != nil {
		return nil, tableConversionUnavailable(err)
	}
	return table, nil
}

// decodeTableRowObjects turns each row's Object.Raw into Object.Object.
// json.Unmarshal only fills Raw; TablePrinter.WithNamespace reads Object via meta.Accessor.
// Same approach as kubectl's decodeIntoTable.
func decodeTableRowObjects(table *metav1.Table) error {
	for i := range table.Rows {
		row := &table.Rows[i]
		if len(row.Object.Raw) == 0 || row.Object.Object != nil {
			continue
		}
		obj, err := runtime.Decode(unstructured.UnstructuredJSONScheme, row.Object.Raw)
		if err != nil {
			return errors.Wrap(err, "decode table row object")
		}
		row.Object.Object = obj
	}
	return nil
}

func newResourceRESTClient(restConfig *rest.Config, gvk schema.GroupVersionKind) (rest.Interface, error) {
	httpClient, err := rest.HTTPClientFor(restConfig)
	if err != nil {
		return nil, err
	}
	return apiutil.RESTClientForGVK(gvk, false, restConfig, scheme.Codecs, httpClient)
}

func labelSelectorString(sel *regionalpb.Selector) string {
	if sel == nil || len(sel.Label) == 0 {
		return ""
	}
	return labels.SelectorFromSet(sel.Label).String()
}

func mapTableRequestError(err error) error {
	if apierrors.IsNotFound(err) {
		return bizerr.New(bizerr.ResourceNotFound, err.Error())
	}
	if statusErr, ok := err.(*apierrors.StatusError); ok && statusErr.Status().Code == http.StatusNotAcceptable {
		return tableConversionUnavailable(err)
	}
	return bizerr.New(bizerr.K8SClientError, errors.Wrap(err, "table list/get").Error())
}

func tableConversionUnavailable(err error) error {
	return bizerr.New(bizerr.ParameterInvalid,
		fmt.Sprintf("table conversion unavailable: %v (hints: try option.detail=true)", err))
}