package server

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

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

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

func newRegistryServer(t *testing.T, name, cmd string) (*Registry, *Migration, *httptest.Server) {
	return newRegistryServerWithExecutor(t, name, fakeRegistryAction(cmd))
}

// newRegistryServerWithExecutor loads one migration whose loop dispatches
// through exec. The retry interval shrinks to test scale so retries fire
// without the production 1-minute wait.
func newRegistryServerWithExecutor(t *testing.T, name string, exec migration.ActionExecutor) (*Registry, *Migration, *httptest.Server) {
	t.Helper()
	migrationsDir := filepath.Join(t.TempDir(), "migrations")
	writeRegistryMigration(t, migrationsDir, name)
	r := NewRegistry(context.Background(), migrationsDir)
	r.loopOptions = []migration.Option{migration.WithActionExecutor(exec)}
	r.retryInterval = time.Millisecond
	r.LoadFromDisk()
	m, ok := r.Get(name)
	require.True(t, ok, "Get(%q) after LoadFromDisk: not found", name)
	srv := httptest.NewServer(RegistryHandler(r, VersionInfo{}))
	t.Cleanup(func() {
		srv.Close()
		m.Cancel(m.Status().JobId)
		waitRegistryIdle(t, m, 5*time.Second)
	})
	return r, m, srv
}

// flakyExecutor counts dispatches; the first failFirst calls return err, the
// rest succeed.
func flakyExecutor(calls *atomic.Int32, failFirst int32, err error) migration.ActionExecutor {
	return func(context.Context, *schema.MigrationLoopPlan, string, *schema.MigrationLoopAction,
		migration.VmctlRunner, migration.ActionOutput,
	) error {
		if calls.Add(1) <= failFirst {
			return err
		}
		return nil
	}
}

func fakeRegistryAction(behavior string) migration.ActionExecutor {
	return func(ctx context.Context, _ *schema.MigrationLoopPlan, _ string, _ *schema.MigrationLoopAction, _ migration.VmctlRunner, output migration.ActionOutput) error {
		if raw, ok := strings.CutPrefix(behavior, "sleep "); ok {
			if _, err := strconv.Atoi(raw); err == nil {
				raw += "s"
			}
			d, err := time.ParseDuration(raw)
			if err != nil {
				return err
			}
			select {
			case <-time.After(d):
				return nil
			case <-ctx.Done():
				return ctx.Err()
			}
		}
		_, err := io.WriteString(output.Log, strings.TrimPrefix(behavior, "echo ")+"\n")
		return err
	}
}

// registryContractCmds returns the four-step command contract keyed by the
// artifact stem.
func registryContractCmds(stem string) []schema.MigrationLoopCmd {
	return []schema.MigrationLoopCmd{
		{Step: "manifest", Action: &schema.MigrationLoopAction{Kind: schema.ActionManifest, TenantsFile: stem + ".json", Manifest: stem + "-manifest.json"}},
		{Step: "migrate", Action: &schema.MigrationLoopAction{Kind: schema.ActionMigrate, Manifest: stem + "-manifest.json", StepInterval: "day"}},
		{Step: "verify", Action: &schema.MigrationLoopAction{Kind: schema.ActionVerify, Manifest: stem + "-manifest.json"}},
	}
}

func writeRegistryMigration(t *testing.T, migrationsDir, name string) {
	t.Helper()
	writeRegistryPlan(t, migrationsDir, name, schema.MigrationLoopPlan{
		Runtime:  &schema.MigrationLoopRuntime{},
		Vmselect: &schema.MigrationLoopVmselect{},
		Tenants: []schema.MigrationLoopTenant{
			{Name: "t1", Cmds: registryContractCmds("t1")},
		},
	})
}

// writeRegistryPlan writes a prepared migration directory with the given
// plan document plus the golden compiled-plan snapshot (the tune views and
// the vmselect jobs derive their specs from it).
func writeRegistryPlan(t *testing.T, migrationsDir, name string, plan schema.MigrationLoopPlan) {
	t.Helper()
	dir := filepath.Join(migrationsDir, name)
	require.NoError(t, os.MkdirAll(dir, 0o755))
	data, err := json.MarshalIndent(plan, "", "  ")
	require.NoError(t, err)
	require.NoError(t, os.WriteFile(filepath.Join(dir, "migration-loop.json"), data, 0o644))
	compiled, err := os.ReadFile(filepath.Join("..", "..", "testdata", "golden", "plan-compile.json"))
	require.NoError(t, err)
	require.NoError(t, os.WriteFile(filepath.Join(dir, "compiled-plan.json"), compiled, 0o644))
}

func waitRegistryIdle(t *testing.T, m *Migration, d time.Duration) {
	waitMigrationIdle(t, m, d)
}

func TestRegistryGet(t *testing.T) {
	t.Run("Status", func(t *testing.T) {
		_, _, srv := newRegistryServer(t, "m1", "echo hi")
		resp, err := http.Get(srv.URL + "/migrations/m1")
		require.NoError(t, err)
		defer resp.Body.Close()
		assert.Equal(t, 200, resp.StatusCode, "status: %d", resp.StatusCode)

		var s migrationStatus
		require.NoError(t, json.NewDecoder(resp.Body).Decode(&s))
		assert.Equal(t, "m1", s.Name)
		assert.False(t, s.Running, "should not be running initially")
	})

	t.Run("UnknownNotFound", func(t *testing.T) {
		_, _, srv := newRegistryServer(t, "m1", "echo hi")
		resp, err := http.Get(srv.URL + "/migrations/no-such")
		require.NoError(t, err)
		defer resp.Body.Close()
		assert.Equal(t, 404, resp.StatusCode, "got %d want 404", resp.StatusCode)
	})
}

func TestRegistryList(t *testing.T) {
	r, _, srv := newRegistryServer(t, "m1", "echo hi")
	writeRegistryMigration(t, r.migrationsDir, "m2")
	r.LoadFromDisk()
	resp, err := http.Get(srv.URL + "/migrations")
	require.NoError(t, err)
	defer resp.Body.Close()

	var body migrationsListResponse
	require.NoError(t, json.NewDecoder(resp.Body).Decode(&body))
	assert.Equal(t, 2, body.Total)
	assert.Len(t, body.Migrations, 2, "List: got %d want 2", len(body.Migrations))
	assert.Equal(t, []string{"m1", "m2"}, []string{body.Migrations[0].Name, body.Migrations[1].Name},
		"migrations must follow registration order")
}

func TestRegistryVersionEndpoint(t *testing.T) {
	v := VersionInfo{Version: "abc1234", Date: "2026-08-31T00:00:00Z"}
	r := NewRegistry(context.Background(), filepath.Join(t.TempDir(), "migrations"))
	srv := httptest.NewServer(RegistryHandler(r, v))
	t.Cleanup(srv.Close)

	resp, err := http.Get(srv.URL + "/version")
	require.NoError(t, err)
	defer resp.Body.Close()
	assert.Equal(t, 200, resp.StatusCode)

	var got VersionInfo
	require.NoError(t, json.NewDecoder(resp.Body).Decode(&got))
	assert.Equal(t, v, got)
}

func TestRegistrySteps(t *testing.T) {
	_, m, srv := newRegistryServer(t, "m1", "echo hi")
	resp, err := http.Get(srv.URL + "/migrations/m1/steps")
	require.NoError(t, err)
	defer resp.Body.Close()
	assert.Equal(t, 200, resp.StatusCode, "steps: got %d want 200", resp.StatusCode)

	var body pendingStepsResponse
	require.NoError(t, json.NewDecoder(resp.Body).Decode(&body))
	require.NotEmpty(t, body.Tenants, "steps should list pending steps")
	assert.Equal(t, "t1", body.Tenants[0].Name)
	assert.False(t, m.Status().Running, "steps must not dispatch a job")
}

func TestRegistryRunAndCancel(t *testing.T) {
	_, m, srv := newRegistryServer(t, "m1", "sleep 30")
	resp, err := http.Post(srv.URL+"/migrations/m1/run", "application/json", nil)
	require.NoError(t, err)
	defer resp.Body.Close()
	assert.Equal(t, 202, resp.StatusCode, "run: got %d want 202", resp.StatusCode)
	assert.True(t, m.Status().Running, "should be running after POST /run")

	cancelResp, err := http.Post(srv.URL+"/migrations/m1/cancel", "application/json", strings.NewReader(`{"jobId":"`+m.Status().JobId+`"}`))
	require.NoError(t, err)
	defer cancelResp.Body.Close()
	assert.Equal(t, 202, cancelResp.StatusCode, "cancel: got %d want 202", cancelResp.StatusCode)
	waitRegistryIdle(t, m, 5*time.Second)
}

func TestRegistryRunSingleFlight(t *testing.T) {
	_, _, srv := newRegistryServer(t, "m1", "sleep 30")
	first, err := http.Post(srv.URL+"/migrations/m1/run", "application/json", nil)
	require.NoError(t, err)
	defer first.Body.Close()
	assert.Equal(t, 202, first.StatusCode, "first: %d", first.StatusCode)

	second, err := http.Post(srv.URL+"/migrations/m1/run", "application/json", nil)
	require.NoError(t, err)
	defer second.Body.Close()
	assert.Equal(t, 409, second.StatusCode, "second run should 409; got %d", second.StatusCode)
}

func TestRegistryDifferentMigrationsRunConcurrently(t *testing.T) {
	r, _, srv := newRegistryServer(t, "m1", "sleep 1")
	writeRegistryMigration(t, r.migrationsDir, "m2")
	r.LoadFromDisk()
	m1, ok := r.Get("m1")
	require.True(t, ok)
	m2, ok := r.Get("m2")
	require.True(t, ok)
	t.Cleanup(func() {
		m2.Cancel(m2.Status().JobId)
		waitRegistryIdle(t, m2, 5*time.Second)
	})

	p1, err := http.Post(srv.URL+"/migrations/m1/run", "application/json", nil)
	require.NoError(t, err)
	defer p1.Body.Close()
	assert.Equal(t, 202, p1.StatusCode, "m1 run: %d", p1.StatusCode)

	p2, err := http.Post(srv.URL+"/migrations/m2/run", "application/json", nil)
	require.NoError(t, err)
	defer p2.Body.Close()
	assert.Equal(t, 202, p2.StatusCode, "m2 run should be 202 (per-migration single-flight): got %d", p2.StatusCode)
	assert.True(t, m1.Status().Running && m2.Status().Running,
		"both migrations should be running concurrently; m1=%v m2=%v", m1.Status().Running, m2.Status().Running)
}

func TestRegistryCancel(t *testing.T) {
	t.Run("WhenIdleConflict", func(t *testing.T) {
		_, _, srv := newRegistryServer(t, "m1", "echo hi")
		resp, err := http.Post(srv.URL+"/migrations/m1/cancel", "application/json", nil)
		require.NoError(t, err)
		defer resp.Body.Close()
		assert.Equal(t, 409, resp.StatusCode, "cancel idle: got %d want 409", resp.StatusCode)
	})

	t.Run("WithMismatchedIdConflict", func(t *testing.T) {
		_, m, srv := newRegistryServer(t, "m1", "sleep 30")
		startResp, err := http.Post(srv.URL+"/migrations/m1/run", "application/json", nil)
		require.NoError(t, err)
		startResp.Body.Close()
		job1Id := m.Status().JobId
		cancelResp, err := http.Post(srv.URL+"/migrations/m1/cancel", "application/json", strings.NewReader(`{"jobId":"`+job1Id+`"}`))
		require.NoError(t, err)
		cancelResp.Body.Close()
		waitRegistryIdle(t, m, 5*time.Second)

		restartResp, err := http.Post(srv.URL+"/migrations/m1/run", "application/json", nil)
		require.NoError(t, err)
		restartResp.Body.Close()
		resp, err := http.Post(srv.URL+"/migrations/m1/cancel", "application/json", strings.NewReader(`{"jobId":"`+job1Id+`"}`))
		require.NoError(t, err)
		defer resp.Body.Close()
		assert.Equal(t, 409, resp.StatusCode, "cancel with stale id: got %d want 409", resp.StatusCode)
	})
}

func TestRegistryLogsStream(t *testing.T) {
	_, m, srv := newRegistryServer(t, "m1", "echo hello-stream")
	startResp, err := http.Post(srv.URL+"/migrations/m1/run", "application/json", nil)
	require.NoError(t, err)
	startResp.Body.Close()
	waitRegistryIdle(t, m, 5*time.Second)

	resp, err := http.Get(srv.URL + "/migrations/m1/logs")
	require.NoError(t, err)
	defer resp.Body.Close()
	ct := resp.Header.Get("Content-Type")
	assert.True(t, strings.HasPrefix(ct, "text/plain"), "Content-Type: got %q want text/plain*", ct)

	body, err := io.ReadAll(resp.Body)
	require.NoError(t, err)
	assert.Contains(t, string(body), "hello-stream", "body missing action output:\n%s", string(body))
	assert.Equal(t, 3, strings.Count(string(body), "hello-stream"), "action output should appear once per step:\n%s", string(body))
	assert.Contains(t, string(body), "=== job finished ===", "body missing done marker:\n%s", string(body))
}

func TestRegistryLogsStreamIdle(t *testing.T) {
	// Before any job has run there is no stream to tail.
	_, _, srv := newRegistryServer(t, "m1", "echo hi")
	resp, err := http.Get(srv.URL + "/migrations/m1/logs")
	require.NoError(t, err)
	defer resp.Body.Close()
	body, err := io.ReadAll(resp.Body)
	require.NoError(t, err)
	assert.Contains(t, string(body), "=== no job running ===", "idle body:\n%s", string(body))
}

func TestRegistrySummary(t *testing.T) {
	t.Run("BeforeRunUnavailable", func(t *testing.T) {
		_, _, srv := newRegistryServer(t, "m1", "echo hi")
		resp, err := http.Get(srv.URL + "/migrations/m1/summary")
		require.NoError(t, err)
		defer resp.Body.Close()
		assert.Equal(t, 503, resp.StatusCode, "got %d want 503", resp.StatusCode)
	})

	t.Run("TextFormat", func(t *testing.T) {
		_, m, srv := newRegistryServer(t, "m1", "echo hi")
		startResp, err := http.Post(srv.URL+"/migrations/m1/run", "application/json", nil)
		require.NoError(t, err)
		startResp.Body.Close()
		waitRegistryIdle(t, m, 5*time.Second)

		resp, err := http.Get(srv.URL + "/migrations/m1/summary?format=text")
		require.NoError(t, err)
		defer resp.Body.Close()
		ct := resp.Header.Get("Content-Type")
		assert.True(t, strings.HasPrefix(ct, "text/plain"), "Content-Type: got %q want text/plain*", ct)

		body, err := io.ReadAll(resp.Body)
		require.NoError(t, err)
		assert.Contains(t, string(body), "total=", "body missing totals line:\n%s", string(body))
		assert.Contains(t, string(body), "tenant", "body missing tenant header:\n%s", string(body))
	})

	t.Run("AfterRun", func(t *testing.T) {
		_, m, srv := newRegistryServer(t, "m1", "echo hi")
		startResp, err := http.Post(srv.URL+"/migrations/m1/run", "application/json", nil)
		require.NoError(t, err)
		startResp.Body.Close()
		waitRegistryIdle(t, m, 5*time.Second)

		resp, err := http.Get(srv.URL + "/migrations/m1/summary")
		require.NoError(t, err)
		defer resp.Body.Close()
		assert.Equal(t, 200, resp.StatusCode, "got %d", resp.StatusCode)

		body, err := io.ReadAll(resp.Body)
		require.NoError(t, err)
		assert.Contains(t, string(body), `"total":`, "body missing total: %s", string(body))
	})
}

func TestRegistryProbeAcceptsEmptyBody(t *testing.T) {
	_, m, srv := newRegistryServer(t, "m1", "echo hi")
	m.plan.Vmselect = nil
	resp, err := http.Post(srv.URL+"/migrations/m1/probe", "application/json", nil)
	require.NoError(t, err)
	defer resp.Body.Close()
	assert.Equal(t, 503, resp.StatusCode, "got %d want 503 (plan missing vmselect)", resp.StatusCode)
}

func TestRegistryStartReturnsJobID(t *testing.T) {
	t.Run("Run", func(t *testing.T) {
		_, m, srv := newRegistryServer(t, "m1", "echo hi")
		resp, err := http.Post(srv.URL+"/migrations/m1/run", "application/json", nil)
		require.NoError(t, err)
		defer resp.Body.Close()
		assert.Equal(t, http.StatusAccepted, resp.StatusCode)

		var body startResponse
		require.NoError(t, json.NewDecoder(resp.Body).Decode(&body))
		assert.True(t, body.Started)
		assert.Regexp(t, `^migrate-run-\d+$`, body.JobId)
		assert.Equal(t, body.JobId, m.Status().JobId, "202 jobId must match the in-flight job")
		waitRegistryIdle(t, m, 5*time.Second)
	})

	t.Run("ExportProbe", func(t *testing.T) {
		_, m, srv := newRegistryServer(t, "m1", "echo hi")
		resp, err := http.Post(srv.URL+"/migrations/m1/probe", "application/json", nil)
		require.NoError(t, err)
		defer resp.Body.Close()
		assert.Equal(t, http.StatusAccepted, resp.StatusCode)

		var body startResponse
		require.NoError(t, json.NewDecoder(resp.Body).Decode(&body))
		assert.True(t, body.Started)
		assert.Regexp(t, `^export-probe-\d+$`, body.JobId)
		assert.Equal(t, body.JobId, m.Status().JobId, "202 jobId must match the in-flight job")
		waitRegistryIdle(t, m, 5*time.Second)
	})
}

func TestRegistryPlan(t *testing.T) {
	_, m, srv := newRegistryServer(t, "m1", "echo hi")

	resp, err := http.Get(srv.URL + "/migrations/m1/plan")
	require.NoError(t, err)
	defer resp.Body.Close()
	assert.Equal(t, 200, resp.StatusCode)
	assert.Equal(t, "application/json", resp.Header.Get("Content-Type"))
	body, err := io.ReadAll(resp.Body)
	require.NoError(t, err)
	onDisk, err := os.ReadFile(m.loopPath())
	require.NoError(t, err)
	assert.Equal(t, onDisk, body, "plan must be served verbatim")

	notFound, err := http.Get(srv.URL + "/migrations/nope/plan")
	require.NoError(t, err)
	defer notFound.Body.Close()
	assert.Equal(t, 404, notFound.StatusCode)
}

func TestRegistryJobsList(t *testing.T) {
	t.Run("Empty", func(t *testing.T) {
		_, _, srv := newRegistryServer(t, "m1", "echo hi")
		resp, err := http.Get(srv.URL + "/migrations/m1/jobs")
		require.NoError(t, err)
		defer resp.Body.Close()
		assert.Equal(t, 200, resp.StatusCode, "got %d want 200", resp.StatusCode)

		var body jobsListResponse
		require.NoError(t, json.NewDecoder(resp.Body).Decode(&body))
		assert.Equal(t, 0, body.Total)
		assert.Empty(t, body.Jobs, "got %d records, want empty", len(body.Jobs))
	})

	t.Run("AfterRun", func(t *testing.T) {
		_, m, srv := newRegistryServer(t, "m1", "echo hi")
		startResp, err := http.Post(srv.URL+"/migrations/m1/run", "application/json", nil)
		require.NoError(t, err)
		startResp.Body.Close()
		waitRegistryIdle(t, m, 5*time.Second)

		resp, err := http.Get(srv.URL + "/migrations/m1/jobs")
		require.NoError(t, err)
		defer resp.Body.Close()

		var body jobsListResponse
		require.NoError(t, json.NewDecoder(resp.Body).Decode(&body))
		assert.Equal(t, 1, body.Total)
		require.Len(t, body.Jobs, 1, "got %d records, want 1", len(body.Jobs))
		assert.Equal(t, jobMigrateRun, body.Jobs[0].Kind)
	})

	t.Run("InflightVisibleByID", func(t *testing.T) {
		_, m, srv := newRegistryServer(t, "m1", "sleep 30")
		startResp, err := http.Post(srv.URL+"/migrations/m1/run", "application/json", nil)
		require.NoError(t, err)
		var body startResponse
		require.NoError(t, json.NewDecoder(startResp.Body).Decode(&body))
		startResp.Body.Close()

		resp, err := http.Get(srv.URL + "/migrations/m1/jobs/" + body.JobId)
		require.NoError(t, err)
		defer resp.Body.Close()
		assert.Equal(t, 200, resp.StatusCode, "in-flight job must be queryable by id")
		var r jobRecord
		require.NoError(t, json.NewDecoder(resp.Body).Decode(&r))
		assert.Equal(t, body.JobId, r.ID)
		assert.Empty(t, r.FinishedAt)

		listResp, err := http.Get(srv.URL + "/migrations/m1/jobs")
		require.NoError(t, err)
		defer listResp.Body.Close()
		var list jobsListResponse
		require.NoError(t, json.NewDecoder(listResp.Body).Decode(&list))
		assert.Equal(t, 1, list.Total, "in-flight job must count toward total")
		assert.Equal(t, body.JobId, list.Jobs[len(list.Jobs)-1].ID, "in-flight job sorts last")

		cancelResp, err := http.Post(srv.URL+"/migrations/m1/cancel", "application/json",
			strings.NewReader(`{"jobId":"`+body.JobId+`"}`))
		require.NoError(t, err)
		cancelResp.Body.Close()
		waitRegistryIdle(t, m, 5*time.Second)
	})
}

func TestRegistryJobGet(t *testing.T) {
	t.Run("ByID", func(t *testing.T) {
		_, m, srv := newRegistryServer(t, "m1", "echo hi")
		startResp, err := http.Post(srv.URL+"/migrations/m1/run", "application/json", nil)
		require.NoError(t, err)
		startResp.Body.Close()
		waitRegistryIdle(t, m, 5*time.Second)
		id := m.Status().LastJobId

		resp, err := http.Get(srv.URL + "/migrations/m1/jobs/" + id)
		require.NoError(t, err)
		defer resp.Body.Close()
		assert.Equal(t, 200, resp.StatusCode, "got %d want 200", resp.StatusCode)

		var r jobRecord
		require.NoError(t, json.NewDecoder(resp.Body).Decode(&r))
		assert.Equal(t, id, r.ID)
	})

	t.Run("UnknownNotFound", func(t *testing.T) {
		_, _, srv := newRegistryServer(t, "m1", "echo hi")
		resp, err := http.Get(srv.URL + "/migrations/m1/jobs/migrate-run-9999999999")
		require.NoError(t, err)
		defer resp.Body.Close()
		assert.Equal(t, 404, resp.StatusCode, "got %d want 404", resp.StatusCode)
	})
}

func TestRegistryRunIgnoresUnknownFields(t *testing.T) {
	_, m, srv := newRegistryServer(t, "m1", "echo hi")
	resp, err := http.Post(
		srv.URL+"/migrations/m1/run",
		"application/json",
		strings.NewReader(`{"retry":true}`),
	)
	require.NoError(t, err)
	require.NoError(t, resp.Body.Close())
	assert.Equal(t, http.StatusAccepted, resp.StatusCode, "status: got %d want %d", resp.StatusCode, http.StatusAccepted)
	waitRegistryIdle(t, m, 5*time.Second)
	jobs := m.Jobs()
	require.Len(t, jobs, 1, "jobs: got %+v, want one migrate-run", jobs)
	assert.Equal(t, jobMigrateRun, jobs[0].Kind)
}

func TestRegistryRunRejectsNegativeRetries(t *testing.T) {
	_, m, srv := newRegistryServer(t, "m1", "echo hi")
	resp, err := http.Post(
		srv.URL+"/migrations/m1/run",
		"application/json",
		strings.NewReader(`{"retries": -1}`),
	)
	require.NoError(t, err)
	defer resp.Body.Close()
	assert.Equal(t, http.StatusBadRequest, resp.StatusCode, "status: got %d want %d", resp.StatusCode, http.StatusBadRequest)
	waitRegistryIdle(t, m, 5*time.Second)
	assert.Empty(t, m.Jobs(), "negative retries must not dispatch a job")
}

func TestRegistryRunWithRetriesFirstTrySuccess(t *testing.T) {
	_, m, srv := newRegistryServer(t, "m1", "echo hi")
	resp, err := http.Post(
		srv.URL+"/migrations/m1/run",
		"application/json",
		strings.NewReader(`{"retries": 3}`),
	)
	require.NoError(t, err)
	require.NoError(t, resp.Body.Close())
	assert.Equal(t, http.StatusAccepted, resp.StatusCode, "status: got %d want %d", resp.StatusCode, http.StatusAccepted)
	waitRegistryIdle(t, m, 5*time.Second)

	jobs := m.Jobs()
	require.Len(t, jobs, 1)
	assert.Empty(t, jobs[0].Result)
	assert.NotContains(t, string(m.logs.Get().snapshot()), "retry", "first-try success must not announce")
}

func TestRegistryRunRetriesRecoverPartialFailure(t *testing.T) {
	var calls atomic.Int32
	_, m, srv := newRegistryServerWithExecutor(t, "m1",
		flakyExecutor(&calls, 1, errors.New("flaky step failure")))
	resp, err := http.Post(
		srv.URL+"/migrations/m1/run",
		"application/json",
		strings.NewReader(`{"retries": 1}`),
	)
	require.NoError(t, err)
	require.NoError(t, resp.Body.Close())
	assert.Equal(t, http.StatusAccepted, resp.StatusCode)
	waitRegistryIdle(t, m, 5*time.Second)

	jobs := m.Jobs()
	require.Len(t, jobs, 1, "one record per request: %+v", jobs)
	assert.Equal(t, jobMigrateRun, jobs[0].Kind)
	assert.Empty(t, jobs[0].Result, "recovered run must report success")
	// Attempt 1 fails the manifest step; attempt 2 runs all three steps.
	assert.Equal(t, int32(4), calls.Load(), "action dispatches")
	assert.Contains(t, string(m.logs.Get().snapshot()), "retry 1/1")
}

func TestRegistryRunRetriesExhaustBudget(t *testing.T) {
	var calls atomic.Int32
	_, m, srv := newRegistryServerWithExecutor(t, "m1",
		flakyExecutor(&calls, 1<<30, errors.New("persistent step failure")))
	resp, err := http.Post(
		srv.URL+"/migrations/m1/run",
		"application/json",
		strings.NewReader(`{"retries": 2}`),
	)
	require.NoError(t, err)
	require.NoError(t, resp.Body.Close())
	assert.Equal(t, http.StatusAccepted, resp.StatusCode)
	waitRegistryIdle(t, m, 5*time.Second)

	jobs := m.Jobs()
	require.Len(t, jobs, 1, "one record per request: %+v", jobs)
	assert.NotEmpty(t, jobs[0].Result, "exhausted run must keep the final error")
	assert.Equal(t, int32(3), calls.Load(), "one manifest dispatch per attempt")
	logs := string(m.logs.Get().snapshot())
	assert.Contains(t, logs, "retry 1/2")
	assert.Contains(t, logs, "retry 2/2")
}

func TestRegistryRunDefaultRetries(t *testing.T) {
	var calls atomic.Int32
	_, m, srv := newRegistryServerWithExecutor(t, "m1",
		flakyExecutor(&calls, 1<<30, errors.New("persistent step failure")))
	resp, err := http.Post(srv.URL+"/migrations/m1/run", "application/json", nil)
	require.NoError(t, err)
	require.NoError(t, resp.Body.Close())
	assert.Equal(t, http.StatusAccepted, resp.StatusCode)
	waitRegistryIdle(t, m, 5*time.Second)

	jobs := m.Jobs()
	require.Len(t, jobs, 1, "one record per request: %+v", jobs)
	assert.NotEmpty(t, jobs[0].Result, "exhausted run must keep the final error")
	assert.Equal(t, int32(DefaultRunRetries+1), calls.Load(), "empty body runs the default budget")
	logs := string(m.logs.Get().snapshot())
	assert.Contains(t, logs, fmt.Sprintf("retry %d/%d", DefaultRunRetries, DefaultRunRetries))
}

func TestRegistryJobArtifact(t *testing.T) {
	t.Run("LoopKindNotFound", func(t *testing.T) {
		_, m, srv := newRegistryServer(t, "m1", "echo hi")
		startResp, err := http.Post(srv.URL+"/migrations/m1/run", "application/json", nil)
		require.NoError(t, err)
		startResp.Body.Close()
		waitRegistryIdle(t, m, 5*time.Second)
		id := m.Status().LastJobId

		resp, err := http.Get(srv.URL + "/migrations/m1/jobs/" + id + "/artifact")
		require.NoError(t, err)
		defer resp.Body.Close()
		assert.Equal(t, 404, resp.StatusCode, "got %d want 404 (loop jobs have no artifact)", resp.StatusCode)
	})

	t.Run("UnknownNotFound", func(t *testing.T) {
		_, _, srv := newRegistryServer(t, "m1", "echo hi")
		resp, err := http.Get(srv.URL + "/migrations/m1/jobs/export-probe-9999999999/artifact")
		require.NoError(t, err)
		defer resp.Body.Close()
		assert.Equal(t, 404, resp.StatusCode, "got %d want 404", resp.StatusCode)
	})

	t.Run("TextFormat", func(t *testing.T) {
		_, m, srv := newRegistryServer(t, "m1", "echo hi")
		rec := &jobRecord{
			ID:         "export-probe-1",
			Kind:       jobExportProbe,
			StartedAt:  "2026-08-21T10:00:00Z",
			FinishedAt: "2026-08-21T10:01:00Z",
		}
		assert.NoError(t, appendJobRecord(m.jobsJSONLPath(), rec))
		art := probeArtifact{
			ID: rec.ID, Kind: "export-probe",
			StartedAt: rec.StartedAt, FinishedAt: rec.FinishedAt,
			Tenants: []probeTenantResult{{
				Tenant: "1001:0",
				Shards: []probeShardRow{{
					ShardID: "1001:0:1",
					Window:  "2026-08-01T00:00:00Z - 2026-08-02T00:00:00Z",
					Bytes:   2048,
					DurMs:   34200,
				}},
			}},
		}
		art.computeTotals()
		artDir := filepath.Join(m.dir, "artifacts")
		require.NoError(t, os.MkdirAll(artDir, 0o755))
		data, err := json.Marshal(art)
		require.NoError(t, err)
		require.NoError(t, os.WriteFile(filepath.Join(artDir, rec.ID+".json"), data, 0o644))

		resp, err := http.Get(srv.URL + "/migrations/m1/jobs/export-probe-1/artifact?format=text")
		require.NoError(t, err)
		defer resp.Body.Close()
		assert.Equal(t, 200, resp.StatusCode)
		assert.Equal(t, "text/plain; charset=utf-8", resp.Header.Get("Content-Type"))
		body, err := io.ReadAll(resp.Body)
		require.NoError(t, err)
		assert.Contains(t, string(body), "1001:0:1")
		assert.Contains(t, string(body), "2026-08-01T00:00:00Z - 2026-08-02T00:00:00Z")
		assert.Contains(t, string(body), "total")

		jsonResp, err := http.Get(srv.URL + "/migrations/m1/jobs/export-probe-1/artifact")
		require.NoError(t, err)
		defer jsonResp.Body.Close()
		assert.Equal(t, "application/json", jsonResp.Header.Get("Content-Type"),
			"default format stays JSON")
	})
}

func TestRegistryTuneAPI(t *testing.T) {
	_, m, srv := newRegistryServer(t, "m1", "echo hi")
	name := m.Name()

	get := func(t *testing.T) (int, string) {
		t.Helper()
		resp, err := http.Get(srv.URL + "/migrations/" + name + "/tune")
		require.NoError(t, err)
		defer resp.Body.Close()
		body, err := io.ReadAll(resp.Body)
		require.NoError(t, err)
		return resp.StatusCode, string(body)
	}
	post := func(t *testing.T, payload string) (int, string) {
		t.Helper()
		resp, err := http.Post(srv.URL+"/migrations/"+name+"/tune", "application/json", strings.NewReader(payload))
		require.NoError(t, err)
		defer resp.Body.Close()
		body, err := io.ReadAll(resp.Body)
		require.NoError(t, err)
		return resp.StatusCode, string(body)
	}

	t.Run("GetDefaultsToPlanValues", func(t *testing.T) {
		code, body := get(t)
		assert.Equal(t, 200, code)
		var view schema.TuneView
		require.NoError(t, json.Unmarshal([]byte(body), &view))
		assert.True(t, view.Vmctl.IsEmpty(), "no plan knobs and no override: vmctl unset")
		assert.Equal(t, "vmselect-migration", view.Vmselect.Migration.Name)
		assert.Equal(t, 1, view.Vmselect.Migration.Replicas)
		assert.Equal(t, "500m", view.Vmselect.Migration.Resources.Requests.CPU)
		assert.Equal(t, "vmselect-candidate", view.Vmselect.Candidate.Name)
		assert.Empty(t, view.Vmselect.UpdatedAt, "no vmselect override")
	})

	t.Run("SetOverridesAndReportsEffective", func(t *testing.T) {
		code, body := post(t, `{"vmctl": {"rateLimitBytes": 20971520, "concurrency": 2}}`)
		assert.Equal(t, 200, code)
		var view schema.TuneView
		require.NoError(t, json.Unmarshal([]byte(body), &view))
		assert.Equal(t, 20971520, *view.Vmctl.RateLimitBytes)
		assert.Equal(t, 2, *view.Vmctl.Concurrency)
		assert.NotEmpty(t, view.Vmctl.UpdatedAt)

		code, body = get(t)
		assert.Equal(t, 200, code)
		require.NoError(t, json.Unmarshal([]byte(body), &view))
		assert.Equal(t, 2, *view.Vmctl.Concurrency)

		_, err := os.Stat(filepath.Join(m.Dir(), "tune.json"))
		assert.NoError(t, err, "override must be persisted next to the plan")
	})

	t.Run("ReplaceIsWholeSectionNotFieldMerge", func(t *testing.T) {
		code, body := post(t, `{"vmctl": {"concurrency": 16}}`)
		assert.Equal(t, 200, code)
		var view schema.TuneView
		require.NoError(t, json.Unmarshal([]byte(body), &view))
		assert.Nil(t, view.Vmctl.RateLimitBytes, "rateLimitBytes reverts to the plan value (unset)")
		assert.Equal(t, 16, *view.Vmctl.Concurrency)
	})

	t.Run("SectionsAreIndependent", func(t *testing.T) {
		code, _ := post(t, `{"vmselect": {"migration": {"replicas": 2}}}`)
		assert.Equal(t, 200, code)
		code, body := post(t, `{"vmctl": {}}`)
		assert.Equal(t, 200, code)
		var view schema.TuneView
		require.NoError(t, json.Unmarshal([]byte(body), &view))
		assert.True(t, view.Vmctl.IsEmpty(), "empty vmctl section clears its override")
		assert.Equal(t, 2, view.Vmselect.Migration.Replicas, "vmselect override untouched")
		assert.NotEmpty(t, view.Vmselect.UpdatedAt)
	})

	t.Run("EmptySectionsClearOverrideAndRemoveFile", func(t *testing.T) {
		code, body := post(t, `{"vmctl": {}, "vmselect": {}}`)
		assert.Equal(t, 200, code)
		var view schema.TuneView
		require.NoError(t, json.Unmarshal([]byte(body), &view))
		assert.True(t, view.Vmctl.IsEmpty())
		assert.Empty(t, view.Vmselect.UpdatedAt)

		_, err := os.Stat(filepath.Join(m.Dir(), "tune.json"))
		assert.True(t, os.IsNotExist(err), "clearing every override removes the file")
	})

	t.Run("InvalidValuesRejected", func(t *testing.T) {
		for _, payload := range []string{
			`{"vmctl": {"concurrency": 0}}`,
			`{"vmctl": {"rateLimitBytes": -1}}`,
			`{"vmselect": {"migration": {"replicas": 0}}}`,
			`{"vmselect": {"migration": {"replicas": 4294967296}}}`,
			`{"vmselect": {"candidate": {"image": ""}}}`,
			`{"vmctl": {"concurrency": 4, "updatedAt": "2020-01-01T00:00:00Z"}}`,
			`{"vmselect": {"migration": {"replicas": 2}, "updatedAt": "2020-01-01T00:00:00Z"}}`,
			`{"unknown": 1}`,
			`not json`,
		} {
			code, body := post(t, payload)
			assert.Equal(t, 422, code, "payload %s: body=%s", payload, body)
		}
	})

	t.Run("UnknownMigration", func(t *testing.T) {
		resp, err := http.Get(srv.URL + "/migrations/nope/tune")
		require.NoError(t, err)
		defer resp.Body.Close()
		assert.Equal(t, 404, resp.StatusCode)
	})
}

func TestRegistryTunePlanKnobsVisible(t *testing.T) {
	migrationsDir := filepath.Join(t.TempDir(), "migrations")
	rate, concurrency := 104857600, 8
	writeRegistryPlan(t, migrationsDir, "m1", schema.MigrationLoopPlan{
		Runtime:  &schema.MigrationLoopRuntime{},
		Vmselect: &schema.MigrationLoopVmselect{},
		Tenants: []schema.MigrationLoopTenant{{Cmds: []schema.MigrationLoopCmd{{
			Step:   "migrate",
			Action: &schema.MigrationLoopAction{Kind: schema.ActionMigrate, RateLimitBytes: &rate, Concurrency: &concurrency},
		}}}},
	})

	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()
	body, err := io.ReadAll(resp.Body)
	require.NoError(t, err)
	assert.Equal(t, 200, resp.StatusCode)
	var view schema.TuneView
	require.NoError(t, json.Unmarshal(body, &view))
	assert.Equal(t, 104857600, *view.Vmctl.RateLimitBytes, "no override: plan knobs are the effective pair")
	assert.Equal(t, 8, *view.Vmctl.Concurrency)
}

func newEmptyRegistryServer(t *testing.T) (*Registry, *httptest.Server) {
	t.Helper()
	r := NewRegistry(context.Background(), filepath.Join(t.TempDir(), "migrations"))
	srv := httptest.NewServer(RegistryHandler(r, VersionInfo{}))
	t.Cleanup(srv.Close)
	return r, srv
}

func TestRegistryPrepareAPI(t *testing.T) {
	r, srv := newEmptyRegistryServer(t)
	compiled := compilePlanBytes(t, []string{"1002:0", "1001:0"})

	resp, err := http.Post(srv.URL+"/migrations/prod-2026-08", "application/json", bytes.NewReader(compiled))
	require.NoError(t, err)
	data, err := io.ReadAll(resp.Body)
	require.NoError(t, err)
	require.NoError(t, resp.Body.Close())
	assert.Equal(t, http.StatusCreated, resp.StatusCode, "body=%s", data)
	var status migrationStatus
	require.NoError(t, json.Unmarshal(data, &status))
	assert.Equal(t, "prod-2026-08", status.Name)

	_, ok := r.Get("prod-2026-08")
	assert.True(t, ok, "prepare must register the migration live")
	_, err = os.Stat(filepath.Join(r.migrationsDir, "prod-2026-08", "migration-loop.json"))
	require.NoError(t, err)

	resp, err = http.Post(srv.URL+"/migrations/prod-2026-08", "application/json", bytes.NewReader(compiled))
	require.NoError(t, err)
	require.NoError(t, resp.Body.Close())
	assert.Equal(t, http.StatusConflict, resp.StatusCode)
}

func TestRegistryPrepareAPIErrors(t *testing.T) {
	cases := []struct {
		name string
		url  string
		body string
		want int
	}{
		{"CollectionPost", "/migrations", `{}`, http.StatusMethodNotAllowed},
		{"InvalidName", "/migrations/FOO", `{}`, http.StatusBadRequest},
		{"GarbagePlan", "/migrations/m1", `{not-json`, http.StatusUnprocessableEntity},
		{"NoTenants", "/migrations/m1", `{"namespace":"x"}`, http.StatusUnprocessableEntity},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			_, srv := newEmptyRegistryServer(t)
			resp, err := http.Post(srv.URL+tc.url, "application/json", strings.NewReader(tc.body))
			require.NoError(t, err)
			require.NoError(t, resp.Body.Close())
			assert.Equal(t, tc.want, resp.StatusCode, "url=%s body=%q", tc.url, tc.body)
		})
	}
}

func TestRegistryPrepareAPIBodyLimit(t *testing.T) {
	_, srv := newEmptyRegistryServer(t)
	orig := maxPrepareBodyBytes
	maxPrepareBodyBytes = 16
	defer func() { maxPrepareBodyBytes = orig }()

	resp, err := http.Post(srv.URL+"/migrations/m1", "application/json",
		strings.NewReader(`{"pad":"`+strings.Repeat("x", 32)+`"}`))
	require.NoError(t, err)
	require.NoError(t, resp.Body.Close())
	assert.Equal(t, http.StatusRequestEntityTooLarge, resp.StatusCode)
}

func TestRegistryStepsMigrateCommands(t *testing.T) {
	migrationsDir := filepath.Join(t.TempDir(), "migrations")
	plan := schema.MigrationLoopPlan{
		Runtime: &schema.MigrationLoopRuntime{
			Region:    "prod",
			Vmcluster: "vmcluster",
			Vminsert:  "http://vminsert/{tenant}/prometheus",
			Window: schema.MigrationLoopWindow{
				Start:        "2026-01-01T00:00:00Z",
				End:          "2026-01-02T00:00:00Z",
				StepInterval: "6h",
			},
		},
		Vmselect: &schema.MigrationLoopVmselect{
			Origin:    "http://o/{tenant}/prometheus",
			Migration: "http://m/{tenant}/prometheus",
			Candidate: "http://c/{tenant}/prometheus",
		},
		Tenants: []schema.MigrationLoopTenant{{Name: "1001:0", TenantsFile: "tenants-1001-0.json", Cmds: registryContractCmds("tenants-1001-0")}},
	}
	writeRegistryPlan(t, migrationsDir, "m1", plan)
	require.NoError(t, os.WriteFile(filepath.Join(migrationsDir, "m1", "tenants-1001-0.json"), []byte(`["1001:0"]`), 0o644))

	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/steps")
	require.NoError(t, err)
	defer resp.Body.Close()
	assert.Equal(t, 200, resp.StatusCode)

	var body pendingStepsResponse
	require.NoError(t, json.NewDecoder(resp.Body).Decode(&body))
	require.Len(t, body.Tenants, 1)
	require.Len(t, body.Tenants[0].Steps, 3)
	migrate := body.Tenants[0].Steps[1]
	assert.Equal(t, "migrate", migrate.Step)
	require.NotEmpty(t, migrate.Cmds, "steps must include vmctl commands before the manifest step runs")
	assert.Contains(t, migrate.Cmds[0], "vmctl vm-native")

	_, err = os.Stat(filepath.Join(migrationsDir, "m1", "tenants-1001-0-manifest.json"))
	assert.True(t, os.IsNotExist(err), "steps must not materialize the manifest")
}