package vmquery

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"testing"
	"time"

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

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

type fakeClient struct {
	responses map[string]string
	calls     []string // baseURL per Query call
}

func (f *fakeClient) Query(ctx context.Context, baseURL, api string, params map[string]string, timeout time.Duration) (*QueryResponse, error) {
	f.calls = append(f.calls, baseURL)
	q := params["query"]
	v, ok := f.responses[q]
	if !ok {
		v = "0"
	}
	body := fmt.Sprintf(`{"status":"success","data":{"resultType":"scalar","result":[1,"%s"]}}`, v)
	var resp QueryResponse
	if err := json.Unmarshal([]byte(body), &resp); err != nil {
		return nil, err
	}
	return &resp, nil
}

func queriedBoth(calls []string) bool {
	var origin, candidate bool
	for _, u := range calls {
		if strings.Contains(u, "vmselect-vmselect-vmcluster") {
			origin = true
		}
		if strings.Contains(u, "vmselect-candidate") {
			candidate = true
		}
	}
	return origin && candidate
}

type differentialClient struct {
	originVal    string
	candidateVal string
}

func (d *differentialClient) Query(ctx context.Context, baseURL, api string, params map[string]string, timeout time.Duration) (*QueryResponse, error) {
	val := d.originVal
	if strings.Contains(baseURL, "vmselect-candidate") {
		val = d.candidateVal
	}
	body := fmt.Sprintf(`{"status":"success","data":{"resultType":"scalar","result":[1,"%s"]}}`, val)
	var resp QueryResponse
	if err := json.Unmarshal([]byte(body), &resp); err != nil {
		return nil, err
	}
	return &resp, nil
}

func writeStatusFixture(t *testing.T, dir string, setImported bool) (manifestPath, statusPath string) {
	t.Helper()
	manifestPath = filepath.Join(dir, "manifest.json")
	manifest := `{
	  "region": "prod", "namespace": "observability", "vmcluster": "vmcluster",
	  "stepInterval": "1d",
	  "tenants": [
	    {"tenant": "1001:0", "sourceAddr": "http://src", "destinationAddr": "http://dst",
	     "shards": [
	       {"shardID": "1001:0:1", "timeStart": "2026-06-16T00:00:00Z", "timeEnd": "2026-06-17T00:00:00Z"}
	     ]}
	  ]
	}`
	require.NoError(t, os.WriteFile(manifestPath, []byte(manifest), 0o644))
	statusPath = status.PathForManifest(manifestPath)
	m, err := status.LoadManifest(manifestPath)
	require.NoError(t, err)
	doc := status.BuildInitialStatus(m, manifestPath)
	if setImported {
		for i := range doc.Shards {
			doc.Shards[i].Phase = schema.PhaseImported
		}
	}
	require.NoError(t, status.WriteJSONAtomic(statusPath, doc))
	return
}

func TestVerifyMarksShardVerified(t *testing.T) {
	dir := t.TempDir()
	manifestPath, statusPath := writeStatusFixture(t, dir, true)

	fake := &fakeClient{
		responses: map[string]string{
			MetricQuery("node_os_info"):                  "100",
			MetricQuery("tidb_executor_statement_total"): "200",
		},
	}
	opts := VerifyOptions{
		Manifest:             manifestPath,
		OriginVmselectURL:    "http://vmselect-vmselect-vmcluster/sel/{tenant}",
		CandidateVmselectURL: "http://vmselect-candidate/sel/{tenant}",
		TimeoutSeconds:       5,
		Tolerance:            0.01,
	}
	var progress bytes.Buffer
	runner, err := NewVerifyRunner(opts, fake, VerifyOutput{Progress: &progress, Log: &progress})
	require.NoError(t, err)
	_, err = runner.Run(context.Background())
	require.NoError(t, err)
	assert.True(t, queriedBoth(fake.calls), "runner should query both origin and candidate vmselect; calls=%v", fake.calls)
	assert.Contains(t, progress.String(), "selected=1", "should select 1 shard")
	assert.Contains(t, progress.String(), "verified=1", "should report verified=1")
	assert.Contains(t, progress.String(), "status=verified origin=k8s=100/tidb=200 candidate=k8s=100/tidb=200 diff=k8s=0/tidb=0",
		"per-shard counts must be visible in progress")

	doc, err := status.LoadDocument(statusPath)
	require.NoError(t, err)
	assert.True(t, doc.Shards[0].Verified, "shard should be verified")
	assert.NotNil(t, doc.Shards[0].Verification, "verification dict should be populated")
}

func TestVerifyFailsOnDiff(t *testing.T) {
	dir := t.TempDir()
	manifestPath, statusPath := writeStatusFixture(t, dir, true)

	client := &differentialClient{originVal: "1000", candidateVal: "100"}
	opts := VerifyOptions{
		Manifest:             manifestPath,
		OriginVmselectURL:    "http://vmselect-vmselect-vmcluster/sel/{tenant}",
		CandidateVmselectURL: "http://vmselect-candidate/sel/{tenant}",
		TimeoutSeconds:       5,
		Tolerance:            0.01,
	}
	var progress bytes.Buffer
	runner, err := NewVerifyRunner(opts, client, VerifyOutput{Progress: &progress, Log: &progress})
	require.NoError(t, err)
	_, err = runner.Run(context.Background())
	assertPartialFailure(t, err)
	assert.Contains(t, progress.String(), "verified=0", "expected verified=0")
	assert.Contains(t, progress.String(), "status=mismatch origin=k8s=1000/tidb=1000 candidate=k8s=100/tidb=100",
		"mismatched counts must be visible in progress")

	doc, err := status.LoadDocument(statusPath)
	require.NoError(t, err)
	assert.False(t, doc.Shards[0].Verified, "shard should not be verified")
}

func assertPartialFailure(t *testing.T, err error) {
	t.Helper()
	assert.Error(t, err)
	var exitErr *exitcode.Error
	require.ErrorAs(t, err, &exitErr)
	assert.Equal(t, exitcode.Partial, exitErr.Code)
}

func TestVerifyStatusWriteFailureFailsRun(t *testing.T) {
	if os.Geteuid() == 0 {
		t.Skip("root bypasses directory permission checks")
	}
	dir := t.TempDir()
	manifestPath, statusPath := writeStatusFixture(t, dir, true)

	require.NoError(t, os.Chmod(dir, 0o500))
	t.Cleanup(func() { _ = os.Chmod(dir, 0o700) })

	fake := &fakeClient{
		responses: map[string]string{
			MetricQuery("node_os_info"):                  "100",
			MetricQuery("tidb_executor_statement_total"): "200",
		},
	}
	opts := VerifyOptions{
		Manifest:             manifestPath,
		OriginVmselectURL:    "http://vmselect-vmselect-vmcluster/sel/{tenant}",
		CandidateVmselectURL: "http://vmselect-candidate/sel/{tenant}",
		TimeoutSeconds:       5,
		Tolerance:            0.01,
	}
	var progress bytes.Buffer
	runner, err := NewVerifyRunner(opts, fake, VerifyOutput{Progress: &progress, Log: &progress})
	require.NoError(t, err)
	_, err = runner.Run(context.Background())
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "persist status for shard", "write failure must fail the pass, not be swallowed")

	// With the parent dir readable again, the on-disk doc must show the last
	// persisted state: nothing verified.
	require.NoError(t, os.Chmod(dir, 0o700))
	doc, err := status.LoadDocument(statusPath)
	require.NoError(t, err)
	assert.False(t, doc.Shards[0].Verified)
}

func TestVerifyRejectsStatusWithNoImportedShards(t *testing.T) {
	dir := t.TempDir()
	manifestPath, _ := writeStatusFixture(t, dir, false)
	opts := VerifyOptions{
		Manifest:             manifestPath,
		OriginVmselectURL:    "http://vmselect-origin/sel/{tenant}",
		CandidateVmselectURL: "http://vmselect-candidate/sel/{tenant}",
		TimeoutSeconds:       5,
		Tolerance:            0.01,
	}
	runner, err := NewVerifyRunner(opts, &fakeClient{}, VerifyOutput{Progress: &bytes.Buffer{}})
	require.NoError(t, err)
	_, err = runner.Run(context.Background())
	assertPartialFailure(t, err)
}

func TestVerifyRejectsPartiallyMigratedStatus(t *testing.T) {
	dir := t.TempDir()
	manifestPath, statusPath := writeStatusFixture(t, dir, true)
	doc, err := status.LoadDocument(statusPath)
	require.NoError(t, err)
	doc.Shards[0].Verified = true
	doc.Shards = append(doc.Shards, schema.Shard{
		ShardID: "1001:0:2",
		Tenant:  "1001:0",
		Phase:   schema.PhasePending,
	})
	require.NoError(t, status.WriteJSONAtomic(statusPath, doc))
	runner, err := NewVerifyRunner(VerifyOptions{
		Manifest:             manifestPath,
		OriginVmselectURL:    "http://vmselect-origin/sel/{tenant}",
		CandidateVmselectURL: "http://vmselect-candidate/sel/{tenant}",
		TimeoutSeconds:       5,
		Tolerance:            0.01,
	}, &fakeClient{}, VerifyOutput{Progress: &bytes.Buffer{}})
	require.NoError(t, err)
	_, err = runner.Run(context.Background())
	assertPartialFailure(t, err)
}

func TestNewRunnerValidation(t *testing.T) {
	cases := []struct {
		name             string
		opts             VerifyOptions
		wantErrSubstring string
	}{
		{
			name:             "ToleranceNegative",
			opts:             VerifyOptions{Manifest: "x", OriginVmselectURL: "o", CandidateVmselectURL: "c", Tolerance: -0.5},
			wantErrSubstring: "tolerance",
		},
		{
			name:             "CandidateVmselectRequired",
			opts:             VerifyOptions{Manifest: "x", OriginVmselectURL: "o", Tolerance: 0.01},
			wantErrSubstring: "candidate-vmselect-url",
		},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			_, err := NewVerifyRunner(tc.opts, nil, VerifyOutput{Progress: &bytes.Buffer{}})
			assert.Error(t, err)
			assert.Contains(t, err.Error(), tc.wantErrSubstring)
		})
	}
}

func TestClassifyError(t *testing.T) {
	f := func(t *testing.T, in, want string) {
		t.Helper()
		assert.Equal(t, want, classifyError(in), "classifyError(%q)", in)
	}
	f(t, "HTTP 503 Service Unavailable", "http")
	f(t, "server returned http 500", "http")
	f(t, "connection refused", "error")
	f(t, "", "error")
}

func TestMutateVerifyFailure(t *testing.T) {
	ts := status.UTCNow()
	doc := &schema.Document{Shards: []schema.Shard{{
		ShardID:      "1001:0:1",
		Verified:     true,
		VerifiedAt:   &ts,
		Verification: &schema.Verification{},
	}}}
	qerr := fmt.Errorf("dial tcp: connection refused")

	require.NoError(t, mutateVerifyFailure(doc, "1001:0:1", qerr))
	s := status.FindShard(doc, "1001:0:1")
	assert.False(t, s.Verified)
	assert.Nil(t, s.VerifiedAt)
	assert.Nil(t, s.Verification)
	if assert.NotNil(t, s.LastError) {
		assert.Equal(t, "dial tcp: connection refused", *s.LastError)
	}

	assert.NoError(t, mutateVerifyFailure(doc, "absent", qerr), "unknown shard must be a no-op")
}