package migration

import (
	"bytes"
	"context"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"os"
	"path/filepath"
	"sync"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	appsv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/client-go/rest"

	"github.com/tidbcloud/observability/vm-migration/internal/retry"
)

// fakeKube is a minimal API server covering the verbs the provisioner uses.
type fakeKube struct {
	mu          sync.Mutex
	deployments map[string]*fakeDeployment
	services    map[string]corev1.Service
	flakyGet    int // respond 500 to the first N deployment GETs
	force409    int // respond 409 to the first N deployment PUTs
	conflicts   int
	terminating bool // DELETE keeps the deployment (foreground propagation)
}

type fakeDeployment struct {
	obj      appsv1.Deployment
	specJSON []byte
	gen      int64
	ready    bool
}

func newFakeKube(t *testing.T) (*fakeKube, *httptest.Server) {
	t.Helper()
	f := &fakeKube{deployments: map[string]*fakeDeployment{}, services: map[string]corev1.Service{}}
	fastPolls(t)
	mux := http.NewServeMux()
	mux.HandleFunc("GET /apis/apps/v1/namespaces/{ns}/deployments/{name}", func(w http.ResponseWriter, r *http.Request) {
		f.mu.Lock()
		defer f.mu.Unlock()
		if f.flakyGet > 0 {
			f.flakyGet--
			w.WriteHeader(http.StatusInternalServerError)
			_, _ = w.Write([]byte(`{"message":"boom"}`))
			return
		}
		d, ok := f.deployments[r.PathValue("ns")+"/"+r.PathValue("name")]
		if !ok {
			w.WriteHeader(http.StatusNotFound)
			_, _ = w.Write([]byte(`{"message":"not found"}`))
			return
		}
		writeJSONForTest(w, http.StatusOK, f.deploymentBody(d))
	})
	mux.HandleFunc("POST /apis/apps/v1/namespaces/{ns}/deployments", func(w http.ResponseWriter, r *http.Request) {
		f.mu.Lock()
		defer f.mu.Unlock()
		var obj appsv1.Deployment
		if err := json.NewDecoder(r.Body).Decode(&obj); err != nil {
			w.WriteHeader(http.StatusBadRequest)
			return
		}
		key := r.PathValue("ns") + "/" + obj.Name
		if _, ok := f.deployments[key]; ok {
			w.WriteHeader(http.StatusConflict)
			_, _ = w.Write([]byte(`{"message":"already exists"}`))
			return
		}
		spec, _ := json.Marshal(obj.Spec)
		f.deployments[key] = &fakeDeployment{obj: obj, specJSON: spec, gen: 1}
		writeJSONForTest(w, http.StatusCreated, f.deploymentBody(f.deployments[key]))
	})
	mux.HandleFunc("PUT /apis/apps/v1/namespaces/{ns}/deployments/{name}", func(w http.ResponseWriter, r *http.Request) {
		f.mu.Lock()
		defer f.mu.Unlock()
		d, ok := f.deployments[r.PathValue("ns")+"/"+r.PathValue("name")]
		if !ok {
			w.WriteHeader(http.StatusNotFound)
			return
		}
		if f.force409 > 0 {
			f.force409--
			f.conflicts++
			w.WriteHeader(http.StatusConflict)
			_, _ = w.Write([]byte(`{"message":"resourceVersion mismatch"}`))
			return
		}
		var obj appsv1.Deployment
		if err := json.NewDecoder(r.Body).Decode(&obj); err != nil {
			w.WriteHeader(http.StatusBadRequest)
			return
		}
		spec, _ := json.Marshal(obj.Spec)
		if !bytes.Equal(spec, d.specJSON) {
			d.gen++
			d.specJSON = spec
		}
		d.obj = obj
		writeJSONForTest(w, http.StatusOK, f.deploymentBody(d))
	})
	mux.HandleFunc("DELETE /apis/apps/v1/namespaces/{ns}/deployments/{name}", func(w http.ResponseWriter, r *http.Request) {
		f.mu.Lock()
		defer f.mu.Unlock()
		key := r.PathValue("ns") + "/" + r.PathValue("name")
		if _, ok := f.deployments[key]; !ok {
			w.WriteHeader(http.StatusNotFound)
			_, _ = w.Write([]byte(`{"message":"not found"}`))
			return
		}
		if !f.terminating {
			delete(f.deployments, key)
		}
		writeJSONForTest(w, http.StatusOK, []byte(`{"kind":"Status","status":"Success"}`))
	})
	mux.HandleFunc("GET /api/v1/namespaces/{ns}/services/{name}", func(w http.ResponseWriter, r *http.Request) {
		f.mu.Lock()
		defer f.mu.Unlock()
		svc, ok := f.services[r.PathValue("ns")+"/"+r.PathValue("name")]
		if !ok {
			w.WriteHeader(http.StatusNotFound)
			_, _ = w.Write([]byte(`{"message":"not found"}`))
			return
		}
		writeJSONForTest(w, http.StatusOK, mustMarshalForTest(svc))
	})
	mux.HandleFunc("POST /api/v1/namespaces/{ns}/services", func(w http.ResponseWriter, r *http.Request) {
		f.mu.Lock()
		defer f.mu.Unlock()
		var obj corev1.Service
		if err := json.NewDecoder(r.Body).Decode(&obj); err != nil {
			w.WriteHeader(http.StatusBadRequest)
			return
		}
		f.services[r.PathValue("ns")+"/"+obj.Name] = obj
		writeJSONForTest(w, http.StatusCreated, mustMarshalForTest(obj))
	})
	mux.HandleFunc("DELETE /api/v1/namespaces/{ns}/services/{name}", func(w http.ResponseWriter, r *http.Request) {
		f.mu.Lock()
		defer f.mu.Unlock()
		key := r.PathValue("ns") + "/" + r.PathValue("name")
		if _, ok := f.services[key]; !ok {
			w.WriteHeader(http.StatusNotFound)
			_, _ = w.Write([]byte(`{"message":"not found"}`))
			return
		}
		delete(f.services, key)
		writeJSONForTest(w, http.StatusOK, []byte(`{"kind":"Status","status":"Success"}`))
	})
	srv := httptest.NewServer(mux)
	t.Cleanup(srv.Close)
	return f, srv
}

// deploymentBody marshals the stored object with server-assigned metadata
// and a synthetic status: the controller observes the generation and reports
// the pod count only once ready flips.
func (f *fakeKube) deploymentBody(d *fakeDeployment) []byte {
	out := d.obj.DeepCopy()
	out.Generation = d.gen
	if d.ready {
		out.Status = appsv1.DeploymentStatus{
			ObservedGeneration: d.gen,
			AvailableReplicas:  derefInt32(out.Spec.Replicas),
		}
	} else {
		out.Status = appsv1.DeploymentStatus{ObservedGeneration: d.gen - 1}
	}
	return mustMarshalForTest(out)
}

func derefInt32(p *int32) int32 {
	if p == nil {
		return 0
	}
	return *p
}

const fakeKey = "observability/" + "vmselect-migration"

func (f *fakeKube) setReady() {
	f.mu.Lock()
	defer f.mu.Unlock()
	if d, ok := f.deployments[fakeKey]; ok {
		d.ready = true
	}
}

// flipReadyWhenPresent marks the deployment ready as soon as it exists.
func (f *fakeKube) flipReadyWhenPresent() {
	for i := 0; i < 1000; i++ {
		f.mu.Lock()
		d, ok := f.deployments[fakeKey]
		if ok {
			d.ready = true
		}
		f.mu.Unlock()
		if ok {
			return
		}
		time.Sleep(2 * time.Millisecond)
	}
}

func (f *fakeKube) deleteDeployment() {
	f.mu.Lock()
	defer f.mu.Unlock()
	delete(f.deployments, fakeKey)
}

func writeJSONForTest(w http.ResponseWriter, status int, body []byte) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(status)
	_, _ = w.Write(body)
}

func mustMarshalForTest(v any) []byte {
	data, err := json.Marshal(v)
	if err != nil {
		panic(err)
	}
	return data
}

// fastPolls shrinks the test-tunable knobs and restores them on cleanup.
func fastPolls(t *testing.T) {
	t.Helper()
	poll, policy := kubePollInterval, kubeRetryPolicy
	kubePollInterval = 10 * time.Millisecond
	kubeRetryPolicy = retry.Policy{MaxAttempts: 3}
	t.Cleanup(func() {
		kubePollInterval, kubeRetryPolicy = poll, policy
	})
}

func newTestClient(t *testing.T, srv *httptest.Server) VmselectProvisioner {
	t.Helper()
	provisioner, err := NewKubeProvisioner(&rest.Config{Host: srv.URL})
	assert.NoError(t, err)
	return provisioner
}

func kubeSpec() VmselectSpec {
	spec := testVmselectSpec()
	spec.StorageIDs = []int{3, 4}
	return spec
}

func TestKubeEnsureCreatesAndWaitsReady(t *testing.T) {
	f, srv := newFakeKube(t)
	c := newTestClient(t, srv)
	go f.flipReadyWhenPresent()

	var progress bytes.Buffer
	action, err := c.Ensure(context.Background(), kubeSpec(), 2*time.Second, &progress)
	assert.NoError(t, err)
	assert.Equal(t, ActionCreated, action)
	assert.Contains(t, progress.String(), "deployment observability/vmselect-migration created")
	assert.Contains(t, progress.String(), "service observability/vmselect-migration created")
	assert.Contains(t, progress.String(), "deployment observability/vmselect-migration ready")
}

func TestKubeEnsureUpdateConverges(t *testing.T) {
	f, srv := newFakeKube(t)
	c := newTestClient(t, srv)
	go f.flipReadyWhenPresent()

	var progress bytes.Buffer
	_, err := c.Ensure(context.Background(), kubeSpec(), 2*time.Second, &progress)
	assert.NoError(t, err)

	changed := kubeSpec()
	changed.Replicas = 2
	progress.Reset()
	action, err := c.Ensure(context.Background(), changed, 2*time.Second, &progress)
	assert.NoError(t, err)
	assert.Equal(t, ActionUpdated, action)
	assert.Contains(t, progress.String(), "deployment observability/vmselect-migration updated")
	assert.Contains(t, progress.String(), "service observability/vmselect-migration present")
}

func TestKubeEnsureConflictRetries(t *testing.T) {
	f, srv := newFakeKube(t)
	c := newTestClient(t, srv)

	// Pre-store a ready deployment; the first PUT hits a forced 409, the
	// re-read and second PUT succeed.
	var progress bytes.Buffer
	go f.flipReadyWhenPresent()
	_, err := c.Ensure(context.Background(), kubeSpec(), 2*time.Second, &progress)
	assert.NoError(t, err)
	f.setReady()

	f.mu.Lock()
	f.force409 = 1
	f.mu.Unlock()
	progress.Reset()
	action, err := c.Ensure(context.Background(), kubeSpec(), 2*time.Second, &progress)
	assert.NoError(t, err)
	assert.Equal(t, ActionUpdated, action)
	assert.Equal(t, 1, f.conflicts)
}

func TestKubeEnsureTransientRetry(t *testing.T) {
	f, srv := newFakeKube(t)
	c := newTestClient(t, srv)
	f.mu.Lock()
	f.flakyGet = 1
	f.mu.Unlock()
	go f.flipReadyWhenPresent()

	var progress bytes.Buffer
	_, err := c.Ensure(context.Background(), kubeSpec(), 2*time.Second, &progress)
	assert.NoError(t, err)
}

func TestKubeEnsureRolloutTimeout(t *testing.T) {
	_, srv := newFakeKube(t)
	c := newTestClient(t, srv)

	var progress bytes.Buffer
	_, err := c.Ensure(context.Background(), kubeSpec(), 60*time.Millisecond, &progress)
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "not ready after")
	assert.Contains(t, progress.String(), "waiting:")
}

func TestKubeEnsureServiceMismatchFails(t *testing.T) {
	f, srv := newFakeKube(t)
	c := newTestClient(t, srv)

	foreign := kubeSpec().Service()
	foreign.Spec.Selector = map[string]string{kubeNameLabel: "something-else"}
	f.mu.Lock()
	f.services[fakeKey] = foreign
	f.mu.Unlock()

	var progress bytes.Buffer
	_, err := c.Ensure(context.Background(), kubeSpec(), 2*time.Second, &progress)
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "unexpected selector or ports")
}

func TestKubeEnsureServiceWithServerDefaultsMatches(t *testing.T) {
	f, srv := newFakeKube(t)
	c := newTestClient(t, srv)

	// The apiserver defaults protocol and assigns clusterIP; those fields
	// must not trip the existing-service validation.
	stored := kubeSpec().Service()
	stored.Spec.Ports[0].Protocol = corev1.ProtocolTCP
	stored.Spec.ClusterIP = "10.0.0.42"
	f.mu.Lock()
	f.services[fakeKey] = stored
	f.mu.Unlock()
	go f.flipReadyWhenPresent()

	var progress bytes.Buffer
	_, err := c.Ensure(context.Background(), kubeSpec(), 2*time.Second, &progress)
	assert.NoError(t, err)
	assert.Contains(t, progress.String(), "service observability/vmselect-migration present")
}

func TestKubeRemovePresentAndAbsent(t *testing.T) {
	f, srv := newFakeKube(t)
	c := newTestClient(t, srv)
	go f.flipReadyWhenPresent()

	var progress bytes.Buffer
	_, err := c.Ensure(context.Background(), kubeSpec(), 2*time.Second, &progress)
	assert.NoError(t, err)

	action, err := c.Remove(context.Background(), "observability", "vmselect-migration", 2*time.Second, &progress)
	assert.NoError(t, err)
	assert.Equal(t, ActionDeleted, action)
	assert.Contains(t, progress.String(), "deployment observability/vmselect-migration deleted")
	assert.Contains(t, progress.String(), "service observability/vmselect-migration deleted")

	progress.Reset()
	action, err = c.Remove(context.Background(), "observability", "vmselect-migration", 2*time.Second, &progress)
	assert.NoError(t, err)
	assert.Equal(t, ActionAbsent, action)
	assert.Contains(t, progress.String(), "deployment observability/vmselect-migration absent")
}

func TestKubeRemoveWaitsForTermination(t *testing.T) {
	f, srv := newFakeKube(t)
	c := newTestClient(t, srv)
	go f.flipReadyWhenPresent()

	var progress bytes.Buffer
	_, err := c.Ensure(context.Background(), kubeSpec(), 2*time.Second, &progress)
	assert.NoError(t, err)

	// Foreground propagation: the object lingers after DELETE until the
	// goroutine releases it.
	f.mu.Lock()
	f.terminating = true
	f.mu.Unlock()
	go func() {
		time.Sleep(40 * time.Millisecond)
		f.deleteDeployment()
	}()

	action, err := c.Remove(context.Background(), "observability", "vmselect-migration", 2*time.Second, &progress)
	assert.NoError(t, err)
	assert.Equal(t, ActionDeleted, action)
}

func TestKubeRemoveTimeout(t *testing.T) {
	f, srv := newFakeKube(t)
	c := newTestClient(t, srv)

	f.mu.Lock()
	f.terminating = true
	f.deployments[fakeKey] = &fakeDeployment{obj: kubeSpec().Deployment()}
	f.mu.Unlock()

	var progress bytes.Buffer
	_, err := c.Remove(context.Background(), "observability", "vmselect-migration", 80*time.Millisecond, &progress)
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "still terminating after")
}

func TestNewKubeProvisionerBearerToken(t *testing.T) {
	tokenFile := filepath.Join(t.TempDir(), "token")
	assert.NoError(t, os.WriteFile(tokenFile, []byte("test-token\n"), 0o600))

	var gotAuth string
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		gotAuth = r.Header.Get("Authorization")
		w.WriteHeader(http.StatusNotFound)
		_, _ = w.Write([]byte(`{"message":"not found"}`))
	}))
	t.Cleanup(srv.Close)

	provisioner, err := NewKubeProvisioner(&rest.Config{Host: srv.URL, BearerTokenFile: tokenFile})
	assert.NoError(t, err)
	_, err = provisioner.Remove(context.Background(), "ns", "name", 100*time.Millisecond, &bytes.Buffer{})
	assert.NoError(t, err)
	assert.Equal(t, "Bearer test-token", gotAuth)
}

func TestNewKubeProvisionerBadCA(t *testing.T) {
	caFile := filepath.Join(t.TempDir(), "ca.crt")
	assert.NoError(t, os.WriteFile(caFile, []byte("not a certificate"), 0o600))
	_, err := NewKubeProvisioner(&rest.Config{Host: "https://example.com", TLSClientConfig: rest.TLSClientConfig{CAFile: caFile}})
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "unable to load root certificates")
}