package cluster

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"net/http"
	"net/http/httptest"
	"testing"

	"github.com/stretchr/testify/require"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client"
	ctrlclientfake "sigs.k8s.io/controller-runtime/pkg/client/fake"
)

func TestWriteK8sResourceResponseStatusCode(t *testing.T) {
	w := httptest.NewRecorder()
	committed, err := writeK8sResourceResponse(w, map[string]bool{"ok": true}, 0)
	require.NoError(t, err)
	require.True(t, committed)
	require.Equal(t, http.StatusOK, w.Code)

	resp := &k8sResourceResponse{}
	err = json.Unmarshal(w.Body.Bytes(), resp)
	require.NoError(t, err)
	require.Equal(t, http.StatusOK, resp.Status)
}

func TestWriteK8sResourceResponseMarshalErrorDoesNotCommitResponse(t *testing.T) {
	w := &trackingResponseWriter{header: make(http.Header)}

	committed, err := writeK8sResourceResponse(w, map[string]interface{}{"bad": make(chan int)}, http.StatusCreated)
	require.Error(t, err)
	require.False(t, committed)
	require.False(t, w.wroteHeader)
	require.Zero(t, w.body.Len())
}

func TestWriteK8sResourceResponseWriteErrorMarksResponseCommitted(t *testing.T) {
	w := &failingWriteResponseWriter{
		trackingResponseWriter: trackingResponseWriter{header: make(http.Header)},
		writeErr:               errors.New("write failed"),
	}

	committed, err := writeK8sResourceResponse(w, map[string]bool{"ok": true}, http.StatusCreated)
	require.Error(t, err)
	require.True(t, committed)
	require.True(t, w.wroteHeader)
	require.Equal(t, http.StatusCreated, w.statusCode)
}

type trackingResponseWriter struct {
	header      http.Header
	body        bytes.Buffer
	statusCode  int
	wroteHeader bool
}

func (w *trackingResponseWriter) Header() http.Header {
	return w.header
}

func (w *trackingResponseWriter) Write(data []byte) (int, error) {
	if !w.wroteHeader {
		w.WriteHeader(http.StatusOK)
	}
	return w.body.Write(data)
}

func (w *trackingResponseWriter) WriteHeader(statusCode int) {
	w.statusCode = statusCode
	w.wroteHeader = true
}

type failingWriteResponseWriter struct {
	trackingResponseWriter
	writeErr error
}

func (w *failingWriteResponseWriter) Write(data []byte) (int, error) {
	return 0, w.writeErr
}

func TestValidateK8sResourceRequest(t *testing.T) {
	t.Run("requires namespace", func(t *testing.T) {
		req := &k8sResourceModel{
			OperateType: operateTypeGet,
			APIVersion:  "cloud.pingcap.com/v1",
			Kind:        "Branch",
			Name:        "br-1",
		}

		err := validateK8sResourceRequest(req)
		require.Error(t, err)
		require.Contains(t, err.Error(), "namespace")
	})

	t.Run("accepts explicit namespace", func(t *testing.T) {
		req := &k8sResourceModel{
			OperateType: operateTypeGet,
			APIVersion:  "cloud.pingcap.com/v1",
			Kind:        "Import",
			Namespace:   "other-ns",
			Name:        "imp-1",
		}

		err := validateK8sResourceRequest(req)
		require.NoError(t, err)
	})

	t.Run("create requires object", func(t *testing.T) {
		req := &k8sResourceModel{
			OperateType: operateTypeCreate,
			APIVersion:  "cloud.pingcap.com/v1",
			Kind:        "Branch",
			Namespace:   "tidb123",
		}

		err := validateK8sResourceRequest(req)
		require.Error(t, err)
		require.Contains(t, err.Error(), "object")
	})

	t.Run("get requires name", func(t *testing.T) {
		req := &k8sResourceModel{
			OperateType: operateTypeGet,
			APIVersion:  "cloud.pingcap.com/v1",
			Kind:        "Branch",
			Namespace:   "tidb123",
		}

		err := validateK8sResourceRequest(req)
		require.Error(t, err)
		require.Contains(t, err.Error(), "name")
	})

	t.Run("update status requires status payload", func(t *testing.T) {
		req := &k8sResourceModel{
			OperateType: operateTypeUpdateStatus,
			APIVersion:  "cloud.pingcap.com/v1",
			Kind:        "Import",
			Namespace:   "tidb123",
			Name:        "imp-1",
		}

		err := validateK8sResourceRequest(req)
		require.Error(t, err)
		require.Contains(t, err.Error(), "status")
	})

	t.Run("requires api version and kind", func(t *testing.T) {
		req := &k8sResourceModel{
			OperateType: operateTypeGet,
			Namespace:   "tidb123",
			Name:        "br-1",
		}

		err := validateK8sResourceRequest(req)
		require.Error(t, err)
		require.Contains(t, err.Error(), "apiVersion")
	})

	t.Run("list does not require name", func(t *testing.T) {
		req := &k8sResourceModel{
			OperateType: operateTypeList,
			APIVersion:  "cloud.pingcap.com/v1",
			Kind:        "Import",
			Namespace:   "tidb123",
		}

		err := validateK8sResourceRequest(req)
		require.NoError(t, err)
	})
}

func TestExecuteK8sResourceOperationGetSplit(t *testing.T) {
	scheme := runtime.NewScheme()
	cacheObj := &unstructured.Unstructured{Object: map[string]interface{}{
		"apiVersion": "cloud.pingcap.com/v1",
		"kind":       "Import",
		"metadata": map[string]interface{}{
			"name":      "imp-1",
			"namespace": "tidb123",
		},
		"spec": map[string]interface{}{
			"source": "cache",
		},
	}}
	directObj := &unstructured.Unstructured{Object: map[string]interface{}{
		"apiVersion": "cloud.pingcap.com/v1",
		"kind":       "Import",
		"metadata": map[string]interface{}{
			"name":      "imp-1",
			"namespace": "tidb123",
		},
		"spec": map[string]interface{}{
			"source": "direct",
		},
	}}

	cacheCli := ctrlclientfake.NewClientBuilder().WithScheme(scheme).WithObjects(cacheObj).Build()
	directCli := ctrlclientfake.NewClientBuilder().WithScheme(scheme).WithObjects(directObj).Build()
	baseReq := &k8sResourceModel{
		APIVersion: "cloud.pingcap.com/v1",
		Kind:       "Import",
		Namespace:  "tidb123",
		Name:       "imp-1",
	}

	getReq := *baseReq
	getReq.OperateType = operateTypeGet
	resp, statusCode, err := executeK8sResourceOperation(context.Background(), cacheCli, directCli, &getReq)
	require.NoError(t, err)
	require.Equal(t, 200, statusCode)
	respObj, ok := resp.(map[string]interface{})
	require.True(t, ok)
	spec, found, err := unstructured.NestedMap(respObj, "spec")
	require.NoError(t, err)
	require.True(t, found)
	require.Equal(t, "cache", spec["source"])

	getDirectlyReq := *baseReq
	getDirectlyReq.OperateType = operateTypeGetDirectly
	resp, statusCode, err = executeK8sResourceOperation(context.Background(), cacheCli, directCli, &getDirectlyReq)
	require.NoError(t, err)
	require.Equal(t, 200, statusCode)
	respObj, ok = resp.(map[string]interface{})
	require.True(t, ok)
	spec, found, err = unstructured.NestedMap(respObj, "spec")
	require.NoError(t, err)
	require.True(t, found)
	require.Equal(t, "direct", spec["source"])
}

func TestExecuteK8sResourceOperationGetNoFallback(t *testing.T) {
	scheme := runtime.NewScheme()
	directObj := &unstructured.Unstructured{Object: map[string]interface{}{
		"apiVersion": "cloud.pingcap.com/v1",
		"kind":       "Import",
		"metadata": map[string]interface{}{
			"name":      "imp-1",
			"namespace": "tidb123",
		},
	}}

	cacheCli := ctrlclientfake.NewClientBuilder().WithScheme(scheme).Build()
	directCli := ctrlclientfake.NewClientBuilder().WithScheme(scheme).WithObjects(directObj).Build()
	req := &k8sResourceModel{
		OperateType: operateTypeGet,
		APIVersion:  "cloud.pingcap.com/v1",
		Kind:        "Import",
		Namespace:   "tidb123",
		Name:        "imp-1",
	}

	_, statusCode, err := executeK8sResourceOperation(context.Background(), cacheCli, directCli, req)
	require.Error(t, err)
	require.Equal(t, 404, statusCode)
	require.True(t, apierrors.IsNotFound(err))
}

func TestExecuteK8sResourceOperationCreateIgnoreAlreadyExists(t *testing.T) {
	scheme := runtime.NewScheme()
	require.NoError(t, corev1.AddToScheme(scheme))
	existingObj := &unstructured.Unstructured{Object: map[string]interface{}{
		"apiVersion": "cloud.pingcap.com/v1",
		"kind":       "Import",
		"metadata": map[string]interface{}{
			"name":      "imp-1",
			"namespace": "tidb123",
		},
	}}

	cacheCli := ctrlclientfake.NewClientBuilder().WithScheme(scheme).Build()
	directCli := ctrlclientfake.NewClientBuilder().WithScheme(scheme).WithObjects(existingObj).Build()
	req := &k8sResourceModel{
		OperateType: operateTypeCreate,
		APIVersion:  "cloud.pingcap.com/v1",
		Kind:        "Import",
		Namespace:   "tidb123",
		Name:        "imp-1",
		Object: map[string]interface{}{
			"apiVersion": "cloud.pingcap.com/v1",
			"kind":       "Import",
			"metadata": map[string]interface{}{
				"name":      "imp-1",
				"namespace": "tidb123",
			},
		},
	}

	resp, statusCode, err := executeK8sResourceOperation(context.Background(), cacheCli, directCli, req)
	require.NoError(t, err)
	require.Equal(t, 200, statusCode)
	require.Nil(t, resp)
}

func TestExecuteK8sResourceOperationCreateEnsuresNamespace(t *testing.T) {
	scheme := runtime.NewScheme()
	require.NoError(t, corev1.AddToScheme(scheme))

	cacheCli := ctrlclientfake.NewClientBuilder().WithScheme(scheme).Build()
	directCli := ctrlclientfake.NewClientBuilder().WithScheme(scheme).Build()
	req := &k8sResourceModel{
		OperateType: operateTypeCreate,
		APIVersion:  "cloud.pingcap.com/v1",
		Kind:        "Import",
		Namespace:   "tidb-ensure-ns",
		Name:        "imp-1",
		Object: map[string]interface{}{
			"apiVersion": "cloud.pingcap.com/v1",
			"kind":       "Import",
			"metadata": map[string]interface{}{
				"name":      "imp-1",
				"namespace": "tidb-ensure-ns",
			},
		},
	}

	resp, statusCode, err := executeK8sResourceOperation(context.Background(), cacheCli, directCli, req)
	require.NoError(t, err)
	require.Equal(t, 200, statusCode)
	require.Nil(t, resp)

	ns := &corev1.Namespace{}
	err = directCli.Get(context.Background(), types.NamespacedName{Name: "tidb-ensure-ns"}, ns)
	require.NoError(t, err)
	require.Equal(t, "tidb-ensure-ns", ns.Name)
}

func TestExecuteK8sResourceOperationDeleteIgnoreNotFound(t *testing.T) {
	scheme := runtime.NewScheme()
	cacheCli := ctrlclientfake.NewClientBuilder().WithScheme(scheme).Build()
	directCli := ctrlclientfake.NewClientBuilder().WithScheme(scheme).Build()
	req := &k8sResourceModel{
		OperateType: operateTypeDelete,
		APIVersion:  "cloud.pingcap.com/v1",
		Kind:        "Import",
		Namespace:   "tidb123",
		Name:        "imp-not-exist",
	}

	resp, statusCode, err := executeK8sResourceOperation(context.Background(), cacheCli, directCli, req)
	require.NoError(t, err)
	require.Equal(t, 200, statusCode)
	require.Nil(t, resp)
}

func TestExecuteK8sResourceOperationUpdateReturnsNoData(t *testing.T) {
	scheme := runtime.NewScheme()
	existingObj := &unstructured.Unstructured{Object: map[string]interface{}{
		"apiVersion": "cloud.pingcap.com/v1",
		"kind":       "Import",
		"metadata": map[string]interface{}{
			"name":            "imp-1",
			"namespace":       "tidb123",
			"resourceVersion": "1",
		},
		"spec": map[string]interface{}{
			"field": "old",
		},
	}}

	cacheCli := ctrlclientfake.NewClientBuilder().WithScheme(scheme).Build()
	directCli := ctrlclientfake.NewClientBuilder().WithScheme(scheme).WithObjects(existingObj).Build()
	req := &k8sResourceModel{
		OperateType: operateTypeUpdate,
		APIVersion:  "cloud.pingcap.com/v1",
		Kind:        "Import",
		Namespace:   "tidb123",
		Name:        "imp-1",
		Object: map[string]interface{}{
			"apiVersion": "cloud.pingcap.com/v1",
			"kind":       "Import",
			"metadata": map[string]interface{}{
				"name":            "imp-1",
				"namespace":       "tidb123",
				"resourceVersion": "1",
			},
			"spec": map[string]interface{}{
				"field": "new",
			},
		},
	}

	resp, statusCode, err := executeK8sResourceOperation(context.Background(), cacheCli, directCli, req)
	require.NoError(t, err)
	require.Equal(t, 200, statusCode)
	require.Nil(t, resp)
}

func TestExecuteK8sResourceOperationUpdateStatusReturnsNoData(t *testing.T) {
	scheme := runtime.NewScheme()
	existingObj := &unstructured.Unstructured{Object: map[string]interface{}{
		"apiVersion": "cloud.pingcap.com/v1",
		"kind":       "Import",
		"metadata": map[string]interface{}{
			"name":      "imp-1",
			"namespace": "tidb123",
		},
		"status": map[string]interface{}{
			"phase": "Pending",
		},
	}}

	cacheCli := ctrlclientfake.NewClientBuilder().WithScheme(scheme).Build()
	directCli := ctrlclientfake.NewClientBuilder().WithScheme(scheme).WithObjects(existingObj).WithStatusSubresource(existingObj).Build()
	req := &k8sResourceModel{
		OperateType: operateTypeUpdateStatus,
		APIVersion:  "cloud.pingcap.com/v1",
		Kind:        "Import",
		Namespace:   "tidb123",
		Name:        "imp-1",
		Status: map[string]interface{}{
			"phase": "Running",
		},
	}

	resp, statusCode, err := executeK8sResourceOperation(context.Background(), cacheCli, directCli, req)
	require.NoError(t, err)
	require.Equal(t, 200, statusCode)
	require.Nil(t, resp)
}

func TestExecuteK8sResourceOperationList(t *testing.T) {
	scheme := runtime.NewScheme()
	cacheObj1 := &unstructured.Unstructured{Object: map[string]interface{}{
		"apiVersion": "cloud.pingcap.com/v1",
		"kind":       "Import",
		"metadata": map[string]interface{}{
			"name":      "imp-1",
			"namespace": "tidb123",
			"labels": map[string]interface{}{
				"app": "demo",
			},
		},
	}}
	cacheObj2 := &unstructured.Unstructured{Object: map[string]interface{}{
		"apiVersion": "cloud.pingcap.com/v1",
		"kind":       "Import",
		"metadata": map[string]interface{}{
			"name":      "imp-2",
			"namespace": "tidb123",
			"labels": map[string]interface{}{
				"app": "other",
			},
		},
	}}

	cacheCli := ctrlclientfake.NewClientBuilder().WithScheme(scheme).WithObjects(cacheObj1, cacheObj2).Build()
	directCli := ctrlclientfake.NewClientBuilder().WithScheme(scheme).Build()
	req := &k8sResourceModel{
		OperateType: operateTypeList,
		APIVersion:  "cloud.pingcap.com/v1",
		Kind:        "Import",
		Namespace:   "tidb123",
		Labels: map[string]string{
			"app": "demo",
		},
	}

	resp, statusCode, err := executeK8sResourceOperation(context.Background(), cacheCli, directCli, req)
	require.NoError(t, err)
	require.Equal(t, 200, statusCode)
	items, ok := resp.([]interface{})
	require.True(t, ok)
	require.Len(t, items, 1)
}

func TestExecuteK8sResourceOperationListInvalidFieldSelector(t *testing.T) {
	scheme := runtime.NewScheme()
	cacheCli := ctrlclientfake.NewClientBuilder().WithScheme(scheme).Build()
	directCli := ctrlclientfake.NewClientBuilder().WithScheme(scheme).Build()
	req := &k8sResourceModel{
		OperateType: operateTypeList,
		APIVersion:  "cloud.pingcap.com/v1",
		Kind:        "Branch",
		Namespace:   "tidb123",
		FieldSelector: map[string]string{
			"status.phase": "",
		},
	}

	_, statusCode, err := executeK8sResourceOperation(context.Background(), cacheCli, directCli, req)
	require.Error(t, err)
	require.Equal(t, 400, statusCode)
	require.Contains(t, err.Error(), "must not be empty")
}

func TestExecuteK8sResourceOperationListUnsupportedFieldSelector(t *testing.T) {
	scheme := runtime.NewScheme()
	cacheCli := ctrlclientfake.NewClientBuilder().WithScheme(scheme).Build()
	directCli := ctrlclientfake.NewClientBuilder().WithScheme(scheme).Build()
	req := &k8sResourceModel{
		OperateType: operateTypeList,
		APIVersion:  "cloud.pingcap.com/v1",
		Kind:        "Import",
		Namespace:   "tidb123",
		FieldSelector: map[string]string{
			"status.phase": "Running",
		},
	}

	_, statusCode, err := executeK8sResourceOperation(context.Background(), cacheCli, directCli, req)
	require.Error(t, err)
	require.Equal(t, 400, statusCode)
	require.Contains(t, err.Error(), "currently only supports")
}

func TestExecuteK8sResourceOperationListBranchPhaseFieldSelector(t *testing.T) {
	scheme := runtime.NewScheme()
	branchRunning := &unstructured.Unstructured{Object: map[string]interface{}{
		"apiVersion": "cloud.pingcap.com/v1",
		"kind":       "Branch",
		"metadata": map[string]interface{}{
			"name":      "br-running",
			"namespace": "tidb123",
		},
		"status": map[string]interface{}{
			"phase": "Running",
		},
	}}
	branchPending := &unstructured.Unstructured{Object: map[string]interface{}{
		"apiVersion": "cloud.pingcap.com/v1",
		"kind":       "Branch",
		"metadata": map[string]interface{}{
			"name":      "br-pending",
			"namespace": "tidb123",
		},
		"status": map[string]interface{}{
			"phase": "Pending",
		},
	}}

	cacheCli := ctrlclientfake.NewClientBuilder().
		WithScheme(scheme).
		WithObjects(branchRunning, branchPending).
		WithIndex(&unstructured.Unstructured{Object: map[string]interface{}{
			"apiVersion": "cloud.pingcap.com/v1",
			"kind":       "Branch",
		}}, "status.phase", func(o client.Object) []string {
			un, ok := o.(*unstructured.Unstructured)
			if !ok {
				return nil
			}
			phase, found, _ := unstructured.NestedString(un.Object, "status", "phase")
			if !found || phase == "" {
				return nil
			}
			return []string{phase}
		}).
		Build()
	directCli := ctrlclientfake.NewClientBuilder().WithScheme(scheme).Build()

	req := &k8sResourceModel{
		OperateType: operateTypeList,
		APIVersion:  "cloud.pingcap.com/v1",
		Kind:        "Branch",
		Namespace:   "tidb123",
		FieldSelector: map[string]string{
			"status.phase": "Running",
		},
	}

	resp, statusCode, err := executeK8sResourceOperation(context.Background(), cacheCli, directCli, req)
	require.NoError(t, err)
	require.Equal(t, 200, statusCode)
	items, ok := resp.([]interface{})
	require.True(t, ok)
	require.Len(t, items, 1)
	item, ok := items[0].(map[string]interface{})
	require.True(t, ok)
	meta, found, err := unstructured.NestedMap(item, "metadata")
	require.NoError(t, err)
	require.True(t, found)
	require.Equal(t, "br-running", meta["name"])
}