package server

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

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"

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

// fakeProvisioner records Ensure/Remove calls and can block on ctx.
type fakeProvisioner struct {
	mu      sync.Mutex
	ensured []migration.VmselectSpec
	removed []string
}

func (f *fakeProvisioner) Ensure(ctx context.Context, spec migration.VmselectSpec, timeout time.Duration, progress io.Writer) (string, error) {
	select {
	case <-ctx.Done():
		return "", ctx.Err()
	case <-time.After(20 * time.Millisecond):
	}
	f.mu.Lock()
	defer f.mu.Unlock()
	f.ensured = append(f.ensured, spec)
	fmt.Fprintf(progress, "ensured %s\n", spec.Name)
	return migration.ActionCreated, nil
}

func (f *fakeProvisioner) Remove(ctx context.Context, namespace, name string, timeout time.Duration, progress io.Writer) (string, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.removed = append(f.removed, namespace+"/"+name)
	fmt.Fprintf(progress, "removed %s\n", name)
	return migration.ActionDeleted, nil
}

func (f *fakeProvisioner) specs() []migration.VmselectSpec {
	f.mu.Lock()
	defer f.mu.Unlock()
	return append([]migration.VmselectSpec(nil), f.ensured...)
}

func (f *fakeProvisioner) removes() []string {
	f.mu.Lock()
	defer f.mu.Unlock()
	return append([]string(nil), f.removed...)
}

func newVmselectRegistryServer(t *testing.T, provisioner migration.VmselectProvisioner) (*Registry, *Migration, *httptest.Server) {
	t.Helper()
	migrationsDir := filepath.Join(t.TempDir(), "migrations")
	writeRegistryMigration(t, migrationsDir, "m1")
	r := NewRegistry(context.Background(), migrationsDir)
	if provisioner != nil {
		r.provisioner = provisioner
	}
	r.LoadFromDisk()
	m, ok := r.Get("m1")
	require.True(t, ok, "Get(m1) after LoadFromDisk: not found")
	srv := httptest.NewServer(RegistryHandler(r, VersionInfo{}))
	t.Cleanup(srv.Close)
	t.Cleanup(func() {
		m.Cancel(m.Status().JobId)
		waitRegistryIdle(t, m, 5*time.Second)
	})
	return r, m, srv
}

func postVmselect(t *testing.T, srv *httptest.Server, path, body string) (int, string) {
	t.Helper()
	resp, err := http.Post(srv.URL+path, "application/json", strings.NewReader(body))
	require.NoError(t, err)
	defer resp.Body.Close()
	data, err := io.ReadAll(resp.Body)
	require.NoError(t, err)
	return resp.StatusCode, string(data)
}

func lastArtifact(t *testing.T, m *Migration) map[string]any {
	t.Helper()
	id := m.Status().LastJobId
	require.NotEmpty(t, id, "no finished job")
	data, err := m.JobArtifact(id)
	require.NoError(t, err)
	var artifact map[string]any
	require.NoError(t, json.Unmarshal(data, &artifact))
	return artifact
}

func TestVmselectDeployAPI(t *testing.T) {
	fake := &fakeProvisioner{}
	_, m, srv := newVmselectRegistryServer(t, fake)

	code, body := postVmselect(t, srv, "/migrations/m1/deploy", "")
	assert.Equal(t, 202, code, body)
	var startResp struct {
		Started bool   `json:"started"`
		JobId   string `json:"jobId"`
	}
	require.NoError(t, json.Unmarshal([]byte(body), &startResp))
	assert.True(t, startResp.Started)
	assert.Contains(t, startResp.JobId, "vmselect-deploy-")
	waitRegistryIdle(t, m, 5*time.Second)

	specs := fake.specs()
	require.Len(t, specs, 2, "empty body deploys both")
	assert.Equal(t, "vmselect-migration", specs[0].Name, "migration deploys first")
	assert.Equal(t, "vmselect-candidate", specs[1].Name)
	assert.Equal(t, []int{3, 4}, specs[0].StorageIDs, "drained ordinals from the compiled plan")
	assert.Equal(t, []int{0, 1, 2}, specs[1].StorageIDs)

	status := m.Status()
	assert.Equal(t, jobVmselectDeploy, status.LastKind)
	assert.Empty(t, status.LastResult)

	artifact := lastArtifact(t, m)
	assert.Equal(t, string(jobVmselectDeploy), artifact["kind"])

	textResp, err := http.Get(srv.URL + "/migrations/m1/jobs/" + m.Status().LastJobId + "/artifact?format=text")
	require.NoError(t, err)
	defer textResp.Body.Close()
	assert.Equal(t, 400, textResp.StatusCode, "text format is export-probe only")
	rows := artifact["vmselects"].([]any)
	require.Len(t, rows, 2)
	first := rows[0].(map[string]any)
	assert.Equal(t, "vmselect-migration", first["name"])
	assert.Equal(t, migration.ActionCreated, first["action"])
	assert.Contains(t, first["image"], "vmselect")
	assert.GreaterOrEqual(t, first["readyAfterSeconds"], float64(0), "readyAfterSeconds is recorded")
}

func TestVmselectDeployMergesTune(t *testing.T) {
	fake := &fakeProvisioner{}
	_, m, srv := newVmselectRegistryServer(t, fake)

	code, body := postVmselect(t, srv, "/migrations/m1/tune", `{"vmselect": {"migration": {"replicas": 3, "image": "vmselect:v9"}}}`)
	assert.Equal(t, 200, code, body)

	code, body = postVmselect(t, srv, "/migrations/m1/deploy", "")
	assert.Equal(t, 202, code, body)
	waitRegistryIdle(t, m, 5*time.Second)

	specs := fake.specs()
	require.Len(t, specs, 2)
	assert.Equal(t, "vmselect-migration", specs[0].Name)
	assert.Equal(t, 3, specs[0].Replicas, "tune override applied at deploy")
	assert.Equal(t, "vmselect:v9", specs[0].Image)

	artifact := lastArtifact(t, m)
	rows := artifact["vmselects"].([]any)
	assert.Equal(t, "vmselect:v9", rows[0].(map[string]any)["image"], "artifact records the applied image")
}

func TestVmselectUndeployAPI(t *testing.T) {
	fake := &fakeProvisioner{}
	_, m, srv := newVmselectRegistryServer(t, fake)

	code, body := postVmselect(t, srv, "/migrations/m1/undeploy", "")
	assert.Equal(t, 202, code, body)
	waitRegistryIdle(t, m, 5*time.Second)

	removed := fake.removes()
	assert.Equal(t, []string{"observability/vmselect-migration", "observability/vmselect-candidate"}, removed)
	assert.Equal(t, jobVmselectUndeploy, m.Status().LastKind)
}

func TestVmselectDeploySingleFlight(t *testing.T) {
	fake := &fakeProvisioner{}
	_, m, srv := newVmselectRegistryServer(t, fake)

	code, _ := postVmselect(t, srv, "/migrations/m1/deploy", "")
	assert.Equal(t, 202, code)

	deadline := time.Now().Add(5 * time.Second)
	for m.Status().JobId == "" && time.Now().Before(deadline) {
		time.Sleep(5 * time.Millisecond)
	}
	require.NotEmpty(t, m.Status().JobId, "first deploy never started")

	code, body := postVmselect(t, srv, "/migrations/m1/deploy", "")
	assert.Equal(t, 409, code, body)
	waitRegistryIdle(t, m, 5*time.Second)
}

func TestTuneRejectsPreVmselectsCompiledPlan(t *testing.T) {
	migrationsDir := filepath.Join(t.TempDir(), "migrations")
	writeRegistryMigration(t, migrationsDir, "m1")
	compiledPath := filepath.Join(migrationsDir, "m1", "compiled-plan.json")
	assert.NoError(t, os.WriteFile(compiledPath, []byte(`{"namespace": "observability", "vmselectDeployment": {"image": "old"}}`), 0o644))

	r := NewRegistry(context.Background(), migrationsDir)
	r.provisioner = &fakeProvisioner{}
	r.LoadFromDisk()
	srv := httptest.NewServer(RegistryHandler(r, VersionInfo{}))
	t.Cleanup(srv.Close)

	resp, err := http.Get(srv.URL + "/migrations/m1/tune")
	require.NoError(t, err)
	defer resp.Body.Close()
	body, err := io.ReadAll(resp.Body)
	require.NoError(t, err)
	assert.Equal(t, 500, resp.StatusCode)
	assert.Contains(t, string(body), "vmselects restructure", "pre-restructure snapshot must surface a recompile instruction")

	code, body2 := postVmselect(t, srv, "/migrations/m1/deploy", "")
	assert.Equal(t, 500, code, body2)
	assert.Contains(t, body2, "vmselects restructure")
}

func TestRejectsBadReplicaSpanCompiledPlan(t *testing.T) {
	migrationsDir := filepath.Join(t.TempDir(), "migrations")
	writeRegistryMigration(t, migrationsDir, "m1")
	compiledPath := filepath.Join(migrationsDir, "m1", "compiled-plan.json")
	assert.NoError(t, os.WriteFile(compiledPath, []byte(`{"namespace": "observability", "targetReplicaCount": 5, "sourceReplicaCount": 3,
		"vmselects": {"image": "img", "migration": {"name": "vmselect-migration"}, "candidate": {"name": "vmselect-candidate"}}}`), 0o644))

	r := NewRegistry(context.Background(), migrationsDir)
	r.provisioner = &fakeProvisioner{}
	r.LoadFromDisk()
	srv := httptest.NewServer(RegistryHandler(r, VersionInfo{}))
	t.Cleanup(srv.Close)

	resp, err := http.Get(srv.URL + "/migrations/m1/tune")
	require.NoError(t, err)
	defer resp.Body.Close()
	body, err := io.ReadAll(resp.Body)
	require.NoError(t, err)
	assert.Equal(t, 500, resp.StatusCode)
	assert.Contains(t, string(body), "replica span", "target >= source must fail closed, not build a negative ordinal range")

	code, body2 := postVmselect(t, srv, "/migrations/m1/deploy", "")
	assert.Equal(t, 500, code, body2)
	assert.Contains(t, body2, "replica span")
}

func TestVmselectDeployWithoutProvisioner(t *testing.T) {
	_, _, srv := newVmselectRegistryServer(t, nil)

	code, body := postVmselect(t, srv, "/migrations/m1/deploy", "")
	assert.Equal(t, 503, code, body)
	assert.Contains(t, body, "in-cluster config unavailable")
}

func TestResumeVmselectJobsNotAutoResumed(t *testing.T) {
	for _, kind := range []jobKind{jobVmselectDeploy, jobVmselectUndeploy} {
		t.Run(string(kind), func(t *testing.T) {
			migrationsDir := filepath.Join(t.TempDir(), "migrations")
			writeRegistryMigration(t, migrationsDir, "m1")
			marker := filepath.Join(migrationsDir, "m1", "jobs-inflight.json")
			assert.NoError(t, os.WriteFile(marker,
				fmt.Appendf(nil, `{"id": "job-1", "kind": %q, "startedAt": "2026-08-22T00:00:00Z"}`, kind), 0o644))

			r := NewRegistry(context.Background(), migrationsDir)
			r.LoadFromDisk()
			resumed := r.ResumeInterrupted()
			assert.Empty(t, resumed, "vmselect jobs must never auto-resume")

			_, err := os.Stat(marker)
			assert.True(t, os.IsNotExist(err), "marker finalized without resume")

			m, ok := r.Get("m1")
			require.True(t, ok)
			assert.False(t, m.Status().Running)
		})
	}
}

func TestTuneWithoutCompiledPlanFails(t *testing.T) {
	migrationsDir := filepath.Join(t.TempDir(), "migrations")
	writeRegistryMigration(t, migrationsDir, "m1")
	require.NoError(t, os.Remove(filepath.Join(migrationsDir, "m1", "compiled-plan.json")))

	r := NewRegistry(context.Background(), migrationsDir)
	r.LoadFromDisk()
	srv := httptest.NewServer(RegistryHandler(r, VersionInfo{}))
	t.Cleanup(srv.Close)

	resp, err := http.Get(srv.URL + "/migrations/m1/tune")
	require.NoError(t, err)
	defer resp.Body.Close()
	assert.Equal(t, 500, resp.StatusCode, "tune views need the compiled-plan snapshot")
}