package topru

import (
	"net/http"
	"net/http/httptest"
	"sort"
	"testing"
	"time"

	"github.com/gin-gonic/gin"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

func TestTopRUTablePath(t *testing.T) {
	got := topRUTablePath("s3://bucket", "org1", "cluster1")
	assert.Equal(t, "s3://bucket/deltalake/org=org1/cluster=cluster1/type=topsql/component=topru/instance=default", got)
}

func TestInitTopRUApi(t *testing.T) {
	gin.SetMode(gin.TestMode)
	router := gin.New()
	group := router.Group("/api/v1")
	InitTopRUApi(group, nil)
	req := httptest.NewRequest(http.MethodGet, "/api/v1/topru/config", nil)
	w := httptest.NewRecorder()
	router.ServeHTTP(w, req)
	assert.Equal(t, http.StatusOK, w.Code)
}

func TestDisplayUserStatsUser(t *testing.T) {
	assert.Equal(t, "alice", displayUserStatsUser("alice"))
	assert.Equal(t, systemMaintenanceUserLabel, displayUserStatsUser("tidb.cloud_admin@%"))
	assert.Equal(t, systemMaintenanceUserLabel, displayUserStatsUser("2Qtwx1e47fNfz4j.zhoupe_chHgYq8R.cloud_admin@%"))
	assert.Equal(t, "", displayUserStatsUser(""))
}

func TestGetUserStatsRequest_Validate(t *testing.T) {
	tests := []struct {
		name    string
		q       GetUserStatsRequest
		wantErr bool
	}{
		{"ok", GetUserStatsRequest{Start: "1", End: "2"}, false},
		{"missing start", GetUserStatsRequest{End: "2"}, true},
		{"missing end", GetUserStatsRequest{Start: "1"}, true},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			err := tt.q.Validate()
			if tt.wantErr {
				assert.Error(t, err)
			} else {
				assert.NoError(t, err)
			}
		})
	}
}

func TestBuildTopRUSummarySQL(t *testing.T) {
	empty := ""
	alice := "alice"
	sql, err := buildTopRUSummarySQL("s3://p", 100, 200, "topru_total_ru", 10, nil, 60)
	assert.NoError(t, err)
	assert.NotContains(t, sql, "user =")
	assert.Contains(t, sql, "GROUP BY sql_digest")
	assert.NotContains(t, sql, "GROUP BY sql_digest, plan_digest")
	assert.Contains(t, sql, "WHERE COALESCE(sql_digest, '') <> ''")
	assert.Contains(t, sql, "LEFT JOIN top_groups g")
	assert.Contains(t, sql, "ON r.sql_digest = g.sql_digest")
	assert.NotContains(t, sql, "g.plan_digest")

	sql, err = buildTopRUSummarySQL("s3://p", 100, 200, "topru_total_ru", 10, &empty, 60)
	assert.NoError(t, err)
	assert.Contains(t, sql, "user = ''")

	sql, err = buildTopRUSummarySQL("s3://p", 100, 200, "topru_total_ru", 10, &alice, 60)
	assert.NoError(t, err)
	assert.Contains(t, sql, "user = 'alice'")

	systemOp := systemMaintenanceUserLabel
	sql, err = buildTopRUSummarySQL("s3://p", 100, 200, "topru_total_ru", 10, &systemOp, 60)
	assert.NoError(t, err)
	assert.Contains(t, sql, "ends_with(COALESCE(user, ''), '.cloud_admin@%')")
	assert.NotContains(t, sql, "user = 'System Operator'")

	assert.Equal(t, "", buildTopRUUserFilter(nil))
	assert.Contains(t, buildTopRUUserFilter(&systemOp), "ends_with")
	assert.Contains(t, buildTopRUUserFilter(&alice), "user = 'alice'")
}

func TestBuildTopRUSummaryPlanSQL(t *testing.T) {
	sql, err := buildTopRUSummaryPlanSQL("s3://p", 100, 200, "topru_total_ru", 10, nil, 60, "")
	require.NoError(t, err)
	assert.Contains(t, sql, "GROUP BY sql_digest, plan_digest")
	assert.NotContains(t, sql, "sql_digest = '")

	sql, err = buildTopRUSummaryPlanSQL("s3://p", 100, 200, "topru_total_ru", 10, nil, 60, "DIGEST'1")
	require.NoError(t, err)
	assert.Contains(t, sql, "sql_digest = 'DIGEST''1'")
	assert.Contains(t, sql, "GROUP BY sql_digest, plan_digest")
}

func TestProcessTopRUSummaryResults_OthersAtEnd(t *testing.T) {
	now := time.Now().UTC()
	sqlA := "sqlA"

	rows := []topruQueryResult{
		{
			SQLDigest:       &sqlA,
			TimeWindow:      &now,
			TotalRU:         100,
			TotalExecCount:  float64Ptr(10),
			TotalDurationNs: float64Ptr(1_000_000),
		},
		{
			SQLDigest:       strPtr(""),
			TimeWindow:      &now,
			TotalRU:         1000,
			TotalExecCount:  float64Ptr(20),
			TotalDurationNs: float64Ptr(2_000_000),
		},
	}

	svc := &Service{}
	resp, err := svc.processTopRUSummaryResults(rows, 60, map[string]string{"sqlA": "SELECT 1"})
	assert.NoError(t, err)
	assert.Len(t, resp.Data, 2)
	assert.Equal(t, "sqlA", resp.Data[0].SQLDigest)
	assert.Equal(t, "SELECT 1", resp.Data[0].SQLText)
	assert.Equal(t, "", resp.Data[1].SQLDigest)
}

func TestProcessTopRUSummaryPlanResults_OthersAtEnd(t *testing.T) {
	now := time.Now().UTC()
	sqlA := "sqlA"
	planA := "planA"

	rows := []topruQueryResult{
		{
			SQLDigest:       &sqlA,
			PlanDigest:      &planA,
			TimeWindow:      &now,
			TotalRU:         100,
			TotalExecCount:  float64Ptr(10),
			TotalDurationNs: float64Ptr(1_000_000),
		},
		{
			SQLDigest:       strPtr(""),
			PlanDigest:      strPtr(""),
			TimeWindow:      &now,
			TotalRU:         1000,
			TotalExecCount:  float64Ptr(20),
			TotalDurationNs: float64Ptr(2_000_000),
		},
	}

	svc := &Service{}
	resp, err := svc.processTopRUSummaryPlanResults(rows, 60, map[string]string{}, map[string]string{"planA": "Point_Get"})
	assert.NoError(t, err)
	assert.Len(t, resp.Data, 2)

	data := append([]TopRUSummaryPlanItem(nil), resp.Data...)
	sort.SliceStable(data, func(i, j int) bool {
		return data[i].SQLDigest < data[j].SQLDigest
	})
	assert.Equal(t, "", data[0].SQLDigest)
	assert.Equal(t, "sqlA", data[1].SQLDigest)
	assert.Equal(t, "Point_Get", data[1].PlanText)

	assert.Equal(t, "sqlA", resp.Data[0].SQLDigest)
	assert.Equal(t, "", resp.Data[1].SQLDigest)
}

func strPtr(v string) *string       { return &v }
func float64Ptr(v float64) *float64 { return &v }

func TestGetSummaryRequest_Validate(t *testing.T) {
	tests := []struct {
		name    string
		q       GetSummaryRequest
		wantErr bool
	}{
		{"ok", GetSummaryRequest{Start: "1", End: "2", Top: "10"}, false},
		{"missing start", GetSummaryRequest{End: "2", Top: "10"}, true},
		{"invalid order_by", GetSummaryRequest{Start: "1", End: "2", Top: "10", OrderBy: "invalid"}, true},
		{"valid order_by ru", GetSummaryRequest{Start: "1", End: "2", Top: "10", OrderBy: "ru"}, false},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			err := tt.q.Validate()
			if tt.wantErr {
				assert.Error(t, err)
			} else {
				assert.NoError(t, err)
			}
		})
	}
}

func TestClusterHeader_Validate(t *testing.T) {
	provider := "aws"
	region := "us-west-2"
	tenantID := "tid"
	clusterID := "cid"
	deployType := TierNextGen
	env := "prod"
	bizType := "biz"
	h := ClusterHeader{
		Provider:   &provider,
		Region:     &region,
		TenantID:   &tenantID,
		ClusterID:  &clusterID,
		DeployType: &deployType,
		Env:        &env,
		BizType:    &bizType,
	}
	assert.NoError(t, h.Validate())
}