package slowlog_explorer

import (
	"bufio"
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"net/http/httptest"
	"strings"
	"testing"
	"time"

	"github.com/gin-gonic/gin"
	"go.opentelemetry.io/otel/trace"
)

func init() {
	gin.SetMode(gin.TestMode)
}

// mockDataProxy starts a local HTTP server that mimics /data-proxy/query.
func mockDataProxy(t *testing.T, listRowCount int) *httptest.Server {
	t.Helper()
	return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if r.URL.Path != "/data-proxy/query" || r.Method != "POST" {
			http.Error(w, "not found", 404)
			return
		}
		var req DataProxyRequest
		json.NewDecoder(r.Body).Decode(&req)
		sqlLower := strings.ToLower(req.SQL)

		// Stats
		if strings.Contains(sqlLower, "approx_percentile") && !strings.Contains(sqlLower, "group by") {
			json.NewEncoder(w).Encode(DataProxyResponse{
				Columns: []string{"total_count", "p50", "p90", "p99", "max_val"},
				Rows:    [][]interface{}{{float64(listRowCount), 0.5, 1.2, 5.0, 100.0}},
			})
			return
		}
		// Top digests
		if strings.Contains(sqlLower, "group by digest") {
			json.NewEncoder(w).Encode(DataProxyResponse{
				Columns: []string{"digest", "cnt", "avg_ms", "p99_ms", "total_ms", "sample_query"},
				Rows: [][]interface{}{
					{"abc123def456", float64(100), 500.0, 2000.0, 50000.0, "SELECT * FROM t"},
					{"789xyz000111", float64(50), 300.0, 1000.0, 15000.0, "INSERT INTO t VALUES (1)"},
				},
			})
			return
		}
		// Trend
		if strings.Contains(sqlLower, "floor(time") {
			json.NewEncoder(w).Encode(DataProxyResponse{
				Columns: []string{"bucket", "val"},
				Rows:    [][]interface{}{{1.7757e9, float64(42)}, {1.77571e9, float64(38)}},
			})
			return
		}
		// Detail
		if strings.Contains(sqlLower, "abs(time") {
			cols := []string{"time", "digest", "plan_digest", "db", "user", "conn_id", "query_time", "query"}
			json.NewEncoder(w).Encode(DataProxyResponse{
				Columns: cols,
				Rows:    [][]interface{}{{1775707648.457, "abc123", "plan456", "test_db", "root", "conn_1", 1.5, "SELECT 1"}},
			})
			return
		}
		// List query — respect LIMIT in SQL
		cols := slowlogColumns
		actualCount := listRowCount
		if idx := strings.Index(sqlLower, "limit "); idx >= 0 {
			limitStr := strings.TrimSpace(sqlLower[idx+6:])
			if n, err := fmt.Sscanf(limitStr, "%d", &actualCount); n == 1 && err == nil {
				if actualCount > listRowCount {
					actualCount = listRowCount
				}
			}
		}
		rows := make([][]interface{}, actualCount)
		for i := range rows {
			row := make([]interface{}, len(cols))
			row[0] = 1775707648.457 - float64(i)*0.5
			row[1] = fmt.Sprintf("digest%04d", i)
			row[2] = fmt.Sprintf("plan%04d", i)
			row[3] = "tidb-0"
			row[4] = "test_db"
			row[5] = "root"
			row[6] = fmt.Sprintf("conn_%d", 1000+i)
			row[7] = "12345"
			row[8] = 1.5 + float64(i)*0.01 // query_time seconds
			for j := 9; j < len(cols)-2; j++ {
				row[j] = float64(0)
			}
			row[20] = float64(1048576)                                         // mem_max
			row[len(cols)-2] = "idx_a"                                         // index_names
			row[len(cols)-1] = fmt.Sprintf("SELECT * FROM t WHERE id = %d", i) // query
			rows[i] = row
		}
		json.NewEncoder(w).Encode(DataProxyResponse{Columns: cols, Rows: rows})
	}))
}

// newTestService creates a Service pointing at the mock data proxy.
func newTestService(mockURL string) *Service {
	return &Service{DataProxyAddr: mockURL}
}

// ---- transformRows ----

func TestTransformRows_Basic(t *testing.T) {
	resp := &DataProxyResponse{
		Columns: []string{"time", "digest", "plan_digest", "db", "user", "conn_id", "query_time", "mem_max", "query"},
		Rows: [][]interface{}{
			{1775707648.457791, "abc123", "plan456", "test_db", "root", "conn_1", 1.5, float64(1048576), "SELECT * FROM users WHERE id = 1"},
			{1775707640.0, "def789", "plan111", "prod_db", "admin", "conn_2", 0.05, float64(2097152), "INSERT INTO logs VALUES (1)"},
		},
	}

	items := transformRows(resp, "cluster123")
	if len(items) != 2 {
		t.Fatalf("expected 2 items, got %d", len(items))
	}

	item := items[0]
	// cluster_id
	if item["cluster_id"] != "cluster123" {
		t.Errorf("cluster_id = %v", item["cluster_id"])
	}
	// event_time is ISO8601 in 2026
	eventTime := fmt.Sprintf("%v", item["event_time"])
	if !strings.Contains(eventTime, "2026") {
		t.Errorf("event_time = %s, want 2026-...", eventTime)
	}
	// event_id format
	eventID := fmt.Sprintf("%v", item["event_id"])
	if !strings.HasPrefix(eventID, "cluster123:abc123:") {
		t.Errorf("event_id = %s", eventID)
	}
	// query_time_ms: 1.5s → 1500ms (int)
	if qt := item["query_time_ms"]; qt != 1500 {
		t.Errorf("query_time_ms = %v (%T), want 1500", qt, qt)
	}
	// mem_max_bytes (int)
	if mb := item["mem_max_bytes"]; mb != 1048576 {
		t.Errorf("mem_max_bytes = %v (%T), want 1048576", mb, mb)
	}
	// query_head
	if qh := item["query_head"]; qh != "SELECT * FROM users WHERE id = 1" {
		t.Errorf("query_head = %v", qh)
	}
	if _, ok := item["_time"]; !ok {
		t.Error("internal _time should be kept before SSE serialization for cursor extraction")
	}
	public := publicItems(items)
	if _, ok := public[0]["_time"]; ok {
		t.Error("SSE public items should not include internal _time")
	}
	// user_name (*string)
	if un, ok := item["user_name"].(*string); !ok || *un != "root" {
		t.Errorf("user_name = %v", item["user_name"])
	}
	t.Logf("OK: event_id=%s time=%s qt_ms=%v mem=%v", eventID, eventTime, item["query_time_ms"], item["mem_max_bytes"])
}

func TestTransformRows_LongQuery(t *testing.T) {
	longQuery := strings.Repeat("SELECT x FROM very_long_table_name WHERE ", 10)
	resp := &DataProxyResponse{
		Columns: []string{"time", "digest", "conn_id", "query"},
		Rows:    [][]interface{}{{1775707648.0, "abc", "c1", longQuery}},
	}
	items := transformRows(resp, "c1")
	qh := fmt.Sprintf("%v", items[0]["query_head"])
	if len(qh) > maxQueryHead {
		t.Errorf("query_head len %d > max %d", len(qh), maxQueryHead)
	}
	if items[0]["query_text"] != longQuery {
		t.Error("query_text should preserve full query")
	}
}

func TestNumericHelpers_ParseRationalStrings(t *testing.T) {
	if got := toFloatOrZero("100092543/100"); got != float64(1000925.43) {
		t.Fatalf("toFloatOrZero rational = %v, want 1000925.43", got)
	}
	if got := toMsFloatOrZero("43075954389/1000000000"); got != float64(43075.95) {
		t.Fatalf("toMsFloatOrZero rational = %v, want 43075.95", got)
	}
	if got := toIntOrZero("72064528/1"); got != 72064528 {
		t.Fatalf("toIntOrZero rational = %v, want 72064528", got)
	}
}

func TestTransformRows_Empty(t *testing.T) {
	items := transformRows(&DataProxyResponse{Columns: []string{"time"}, Rows: nil}, "c1")
	if items != nil {
		t.Errorf("expected nil for empty rows, got %d items", len(items))
	}
}

func TestParseTimeRangeRejectsInvalidInput(t *testing.T) {
	tests := []string{
		"/slowlogs?start_time=bad&end_time=2026-04-09T03:00:00Z",
		"/slowlogs?start_time=2026-04-09T03:00:00Z&end_time=2026-04-09T00:00:00Z",
	}
	for _, target := range tests {
		w := httptest.NewRecorder()
		c, _ := gin.CreateTestContext(w)
		c.Request = httptest.NewRequest("GET", target, nil)

		if _, _, ok := parseTimeRange(c); ok {
			t.Fatalf("parseTimeRange(%s) should fail", target)
		}
		if w.Code != http.StatusBadRequest {
			t.Fatalf("status = %d, want %d for %s", w.Code, http.StatusBadRequest, target)
		}
	}
}

func TestParseDetailParamsRejectsInvalidDigest(t *testing.T) {
	w := httptest.NewRecorder()
	c, _ := gin.CreateTestContext(w)
	c.Request = httptest.NewRequest("GET", "/detail?digest=not-a-hex&time=1775707648.457&conn_id=conn_1", nil)

	if _, _, _, ok := parseDetailParams(c); ok {
		t.Fatal("parseDetailParams should reject non-hex digest")
	}
	if w.Code != http.StatusBadRequest {
		t.Fatalf("status = %d, want %d", w.Code, http.StatusBadRequest)
	}
}

func TestSelectSlowlogColumnsForMaxCompute(t *testing.T) {
	fields := "query_time, db,query_time,time"
	got, err := selectSlowlogColumns(&fields, slowlogQueryDialectMaxCompute, "time")
	if err != nil {
		t.Fatalf("selectSlowlogColumns: %v", err)
	}

	want := "query_time,db,time,digest,conn_id"
	if joined := strings.Join(got, ","); joined != want {
		t.Fatalf("selected columns = %q, want %q", joined, want)
	}
}

func TestSelectSlowlogColumnsRejectsUnknownField(t *testing.T) {
	fields := "db,not_a_slowlog_column"
	if _, err := selectSlowlogColumns(&fields, slowlogQueryDialectMaxCompute, "time"); err == nil {
		t.Fatal("selectSlowlogColumns should reject an unknown field")
	}
}

func TestSelectSlowlogColumnsRejectsExplicitEmptyMaxComputeFields(t *testing.T) {
	for _, fields := range []string{"  ", "db,"} {
		if _, err := selectSlowlogColumns(&fields, slowlogQueryDialectMaxCompute, "time"); err == nil {
			t.Fatalf("selectSlowlogColumns should reject invalid selection %q", fields)
		}
	}
}

func TestSelectSlowlogColumnsKeepsRequestedQuery(t *testing.T) {
	fields := "query,db"
	got, err := selectSlowlogColumns(&fields, slowlogQueryDialectMaxCompute, "time")
	if err != nil {
		t.Fatalf("selectSlowlogColumns: %v", err)
	}
	if joined, want := strings.Join(got, ","), "query,db,time,digest,conn_id"; joined != want {
		t.Fatalf("selected columns = %q, want %q", joined, want)
	}
}

func TestSelectSlowlogColumnsIncludesHiddenOrderByColumn(t *testing.T) {
	fields := "db"
	got, err := selectSlowlogColumns(&fields, slowlogQueryDialectMaxCompute, "query_time")
	if err != nil {
		t.Fatalf("selectSlowlogColumns: %v", err)
	}
	if joined, want := strings.Join(got, ","), "db,time,digest,conn_id,query_time"; joined != want {
		t.Fatalf("selected columns = %q, want %q", joined, want)
	}
}

func TestSelectSlowlogColumnsKeepsFullProjectionByDefault(t *testing.T) {
	tests := []struct {
		name    string
		fields  *string
		dialect slowlogQueryDialect
	}{
		{name: "MaxCompute without fields", dialect: slowlogQueryDialectMaxCompute},
		{name: "Athena ignores invalid fields", fields: stringPointer("not_a_slowlog_column"), dialect: slowlogQueryDialectDefault},
		{name: "BigQuery ignores invalid fields", fields: stringPointer("not_a_slowlog_column"), dialect: slowlogQueryDialectBigQuery},
		{name: "Synapse ignores invalid fields", fields: stringPointer("not_a_slowlog_column"), dialect: slowlogQueryDialectSynapse},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got, err := selectSlowlogColumns(tt.fields, tt.dialect, "query_time")
			if err != nil {
				t.Fatalf("selectSlowlogColumns: %v", err)
			}
			if strings.Join(got, ",") != strings.Join(slowlogColumns, ",") {
				t.Fatalf("selected %d columns, want the full %d-column projection", len(got), len(slowlogColumns))
			}
		})
	}
}

func TestSlowlogExplorerTraceID(t *testing.T) {
	t.Run("header fallback", func(t *testing.T) {
		c, _ := gin.CreateTestContext(httptest.NewRecorder())
		c.Request = httptest.NewRequest(http.MethodGet, "/slowlogs", nil)
		c.Request.Header.Set("X-Trace-ID", "fallback-trace-id")
		if got := slowlogExplorerTraceID(c); got != "fallback-trace-id" {
			t.Fatalf("trace ID = %q, want header fallback", got)
		}
	})

	t.Run("span takes precedence", func(t *testing.T) {
		traceID := trace.TraceID{1, 2, 3}
		spanContext := trace.NewSpanContext(trace.SpanContextConfig{
			TraceID: traceID,
			SpanID:  trace.SpanID{1},
		})
		request := httptest.NewRequest(http.MethodGet, "/slowlogs", nil)
		request.Header.Set("X-Trace-ID", "fallback-trace-id")
		request = request.WithContext(trace.ContextWithSpanContext(context.Background(), spanContext))
		c, _ := gin.CreateTestContext(httptest.NewRecorder())
		c.Request = request
		if got := slowlogExplorerTraceID(c); got != traceID.String() {
			t.Fatalf("trace ID = %q, want span trace ID %q", got, traceID.String())
		}
	})
}

func stringPointer(value string) *string {
	return &value
}

// ---- SSE format ----

type sseEvent struct {
	Type string
	Data map[string]interface{}
}

func parseSSEEvents(body string) []sseEvent {
	var events []sseEvent
	scanner := bufio.NewScanner(strings.NewReader(body))
	var eventType, eventData string
	for scanner.Scan() {
		line := scanner.Text()
		if strings.HasPrefix(line, "event: ") {
			eventType = strings.TrimPrefix(line, "event: ")
		} else if strings.HasPrefix(line, "data: ") {
			eventData = strings.TrimPrefix(line, "data: ")
		} else if line == "" && eventType != "" {
			var data map[string]interface{}
			json.Unmarshal([]byte(eventData), &data)
			events = append(events, sseEvent{Type: eventType, Data: data})
			eventType, eventData = "", ""
		}
	}
	return events
}

func TestWriteSSECommentFlushesImmediately(t *testing.T) {
	w := httptest.NewRecorder()
	c, _ := gin.CreateTestContext(w)
	c.Request = httptest.NewRequest(http.MethodGet, "/slowlogs", nil)

	writeSSEComment(c, "stream opened")

	if !w.Flushed {
		t.Fatal("SSE comment was not flushed")
	}
	if got, want := w.Body.String(), ": stream opened\n\n"; got != want {
		t.Fatalf("SSE comment = %q, want %q", got, want)
	}
}

func TestWaitForSlowlogQueryWritesHeartbeat(t *testing.T) {
	w := httptest.NewRecorder()
	c, _ := gin.CreateTestContext(w)
	c.Request = httptest.NewRequest(http.MethodGet, "/slowlogs", nil)
	want := &DataProxyResponse{Columns: []string{"time"}}

	got, err := waitForSlowlogQuery(c, 5*time.Millisecond, func(ctx context.Context) (*DataProxyResponse, error) {
		timer := time.NewTimer(30 * time.Millisecond)
		defer timer.Stop()
		select {
		case <-timer.C:
			return want, nil
		case <-ctx.Done():
			return nil, ctx.Err()
		}
	})
	if err != nil {
		t.Fatalf("waitForSlowlogQuery: %v", err)
	}
	if got != want {
		t.Fatal("waitForSlowlogQuery returned an unexpected response")
	}
	if !strings.Contains(w.Body.String(), ": heartbeat\n\n") {
		t.Fatalf("missing heartbeat comment in %q", w.Body.String())
	}
	if !w.Flushed {
		t.Fatal("heartbeat was not flushed")
	}
}

func TestWaitForSlowlogQueryStopsOnRequestCancellation(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	w := httptest.NewRecorder()
	c, _ := gin.CreateTestContext(w)
	c.Request = httptest.NewRequest(http.MethodGet, "/slowlogs", nil).WithContext(ctx)
	queryStopped := make(chan struct{})

	_, err := waitForSlowlogQuery(c, time.Hour, func(ctx context.Context) (*DataProxyResponse, error) {
		cancel()
		<-ctx.Done()
		close(queryStopped)
		return nil, ctx.Err()
	})
	if err != context.Canceled {
		t.Fatalf("waitForSlowlogQuery error = %v, want context.Canceled", err)
	}
	select {
	case <-queryStopped:
	case <-time.After(time.Second):
		t.Fatal("query did not observe request cancellation")
	}
	if w.Body.Len() != 0 {
		t.Fatalf("wrote SSE data after cancellation: %q", w.Body.String())
	}
}

func TestWaitForSlowlogQueryReturnsQueryError(t *testing.T) {
	w := httptest.NewRecorder()
	c, _ := gin.CreateTestContext(w)
	c.Request = httptest.NewRequest(http.MethodGet, "/slowlogs", nil)
	wantErr := fmt.Errorf("query failed")

	result, err := waitForSlowlogQuery(c, time.Hour, func(context.Context) (*DataProxyResponse, error) {
		return nil, wantErr
	})
	if result != nil {
		t.Fatalf("result = %#v, want nil", result)
	}
	if err != wantErr {
		t.Fatalf("error = %v, want %v", err, wantErr)
	}
	if w.Body.Len() != 0 {
		t.Fatalf("wrote unexpected SSE data: %q", w.Body.String())
	}
}

func TestSSE_SmallBatch(t *testing.T) {
	// 6 rows, limit=5: single batch, has_more=true
	mock := mockDataProxy(t, 6)
	defer mock.Close()
	svc := newTestService(mock.URL)

	start, _ := time.Parse(time.RFC3339, "2026-04-09T00:00:00Z")
	end, _ := time.Parse(time.RFC3339, "2026-04-09T03:00:00Z")

	sql := buildSlowlogSQL(start, end, 5+1, nil, nil, "time DESC")
	result, err := svc.queryDataProxy(context.Background(), "c1", sql, 60)
	if err != nil {
		t.Fatal(err)
	}
	items := transformRows(result, "c1")
	hasMore := len(items) > 5
	if hasMore {
		items = items[:5]
	}

	w := httptest.NewRecorder()
	fmt.Fprintf(w, "event: batch\ndata: ")
	data, _ := json.Marshal(map[string]interface{}{"items": items, "batch_index": 0, "is_last": true})
	fmt.Fprintf(w, "%s\n\n", data)
	data, _ = json.Marshal(map[string]interface{}{"total_rows": len(items), "next_cursor": extractLastTime(items), "has_more": hasMore})
	fmt.Fprintf(w, "event: done\ndata: %s\n\n", data)

	events := parseSSEEvents(w.Body.String())
	if len(events) != 2 {
		t.Fatalf("expected 2 events, got %d", len(events))
	}
	if events[0].Type != "batch" || events[1].Type != "done" {
		t.Errorf("types: %s, %s", events[0].Type, events[1].Type)
	}
	batchItems := events[0].Data["items"].([]interface{})
	if len(batchItems) != 5 {
		t.Errorf("batch items: %d, want 5", len(batchItems))
	}
	if events[0].Data["is_last"] != true {
		t.Error("is_last should be true")
	}
	if events[1].Data["has_more"] != true {
		t.Error("has_more should be true")
	}
	if events[1].Data["next_cursor"] == nil {
		t.Error("next_cursor should not be nil")
	}
	t.Logf("SmallBatch OK: 5 items, has_more=true, cursor=%v", events[1].Data["next_cursor"])
}

func TestSSE_TwoPhase(t *testing.T) {
	// 51 rows total, limit=50: two-phase, has_more=true
	mock := mockDataProxy(t, 51)
	defer mock.Close()
	svc := newTestService(mock.URL)

	start, _ := time.Parse(time.RFC3339, "2026-04-09T00:00:00Z")
	end, _ := time.Parse(time.RFC3339, "2026-04-09T03:00:00Z")

	// Phase 1
	sql1 := buildSlowlogSQL(start, end, streamFirstBatch, nil, nil, "time DESC")
	result1, _ := svc.queryDataProxy(context.Background(), "c1", sql1, 60)
	items1 := transformRows(result1, "c1")

	// Phase 2
	sql2 := buildSlowlogSQL(start, end, 50+1, nil, nil, "time DESC")
	result2, _ := svc.queryDataProxy(context.Background(), "c1", sql2, 120)
	allItems := transformRows(result2, "c1")
	hasMore := len(allItems) > 50
	if hasMore {
		allItems = allItems[:50]
	}
	var items2 []map[string]interface{}
	if len(allItems) > streamFirstBatch {
		items2 = allItems[streamFirstBatch:]
	}

	// Build SSE
	w := httptest.NewRecorder()
	writeBatch := func(items []map[string]interface{}, idx int, last bool) {
		d, _ := json.Marshal(map[string]interface{}{"items": items, "batch_index": idx, "is_last": last})
		fmt.Fprintf(w, "event: batch\ndata: %s\n\n", d)
	}
	writeDone := func(total int, cursor *float64, more bool) {
		d, _ := json.Marshal(map[string]interface{}{"total_rows": total, "next_cursor": cursor, "has_more": more})
		fmt.Fprintf(w, "event: done\ndata: %s\n\n", d)
	}

	writeBatch(items1, 0, false)
	writeBatch(items2, 1, true)
	writeDone(len(allItems), extractLastTime(allItems), hasMore)

	events := parseSSEEvents(w.Body.String())
	if len(events) != 3 {
		t.Fatalf("expected 3 events, got %d", len(events))
	}

	b0 := events[0].Data["items"].([]interface{})
	b1 := events[1].Data["items"].([]interface{})
	if len(b0) != 30 {
		t.Errorf("batch0: %d items, want 30", len(b0))
	}
	if events[0].Data["is_last"] != false {
		t.Error("batch0 is_last should be false")
	}
	if len(b1) != 20 {
		t.Errorf("batch1: %d items, want 20", len(b1))
	}
	if events[1].Data["is_last"] != true {
		t.Error("batch1 is_last should be true")
	}
	if events[2].Data["total_rows"] != float64(50) {
		t.Errorf("total_rows = %v, want 50", events[2].Data["total_rows"])
	}
	if events[2].Data["has_more"] != true {
		t.Error("has_more should be true")
	}
	t.Logf("TwoPhase OK: batch0=%d, batch1=%d, total=50, has_more=true", len(b0), len(b1))
}

func TestSSE_NoMore(t *testing.T) {
	// 3 rows, limit=5: single batch, has_more=false
	mock := mockDataProxy(t, 3)
	defer mock.Close()
	svc := newTestService(mock.URL)

	start, _ := time.Parse(time.RFC3339, "2026-04-09T00:00:00Z")
	end, _ := time.Parse(time.RFC3339, "2026-04-09T03:00:00Z")
	sql := buildSlowlogSQL(start, end, 5+1, nil, nil, "time DESC")
	result, _ := svc.queryDataProxy(context.Background(), "c1", sql, 60)
	items := transformRows(result, "c1")
	hasMore := len(items) > 5
	// 3 < 6, so hasMore=false
	if hasMore {
		t.Error("should not have more with only 3 rows")
	}
	t.Logf("NoMore OK: %d items, has_more=%v", len(items), hasMore)
}

// ---- Handler-level tests via mock data proxy ----

func TestHandler_Stats(t *testing.T) {
	mock := mockDataProxy(t, 42)
	defer mock.Close()
	svc := newTestService(mock.URL)

	w := httptest.NewRecorder()
	c, _ := gin.CreateTestContext(w)
	c.Request = httptest.NewRequest("GET", "/stats?start_time=2026-04-09T00:00:00Z&end_time=2026-04-09T03:00:00Z", nil)

	// Call stats logic directly (bypass CanReadCluster)
	startTime, endTime, ok := parseTimeRange(c)
	if !ok {
		t.Fatal("parseTimeRange failed")
	}
	filterClauses, _ := parseFilters(c.Query("filters"))
	sql := buildStatsSQL(startTime, endTime, filterClauses)
	result, err := svc.queryDataProxy(c.Request.Context(), "test-cluster", sql, 60)
	if err != nil {
		t.Fatal(err)
	}
	if len(result.Rows) == 0 {
		t.Fatal("no rows")
	}
	row := result.Rows[0]
	resp := gin.H{
		"total_count": toInt(rawAt(row, 0)),
		"p50_ms":      toMsFloat(rawAt(row, 1)),
		"p90_ms":      toMsFloat(rawAt(row, 2)),
		"p99_ms":      toMsFloat(rawAt(row, 3)),
		"max_ms":      toMsFloat(rawAt(row, 4)),
	}
	c.JSON(http.StatusOK, resp)

	if w.Code != 200 {
		t.Fatalf("status %d", w.Code)
	}
	var body map[string]interface{}
	json.Unmarshal(w.Body.Bytes(), &body)
	if body["total_count"] != float64(42) {
		t.Errorf("total_count = %v, want 42", body["total_count"])
	}
	if body["p50_ms"] != float64(500) {
		t.Errorf("p50_ms = %v, want 500", body["p50_ms"])
	}
	t.Logf("Stats OK: %v", body)
}

func TestHandler_TopDigests(t *testing.T) {
	mock := mockDataProxy(t, 0)
	defer mock.Close()
	svc := newTestService(mock.URL)

	w := httptest.NewRecorder()
	c, _ := gin.CreateTestContext(w)
	c.Request = httptest.NewRequest("GET", "/top-digests?start_time=2026-04-09T00:00:00Z&end_time=2026-04-09T03:00:00Z", nil)

	startTime, endTime, ok := parseTimeRange(c)
	if !ok {
		t.Fatal("parseTimeRange failed")
	}
	sql := buildTopDigestsSQL(startTime, endTime, 5, nil)
	result, err := svc.queryDataProxy(c.Request.Context(), "test-cluster", sql, 60)
	if err != nil {
		t.Fatal(err)
	}
	items := make([]map[string]interface{}, 0)
	for _, row := range result.Rows {
		items = append(items, map[string]interface{}{
			"digest": rawStr(row, 0), "count": toInt(rawAt(row, 1)),
			"avg_ms": toFloat(rawAt(row, 2)), "p99_ms": toFloat(rawAt(row, 3)),
			"total_ms": toFloat(rawAt(row, 4)), "sample_query": rawStr(row, 5),
		})
	}
	c.JSON(http.StatusOK, gin.H{"items": items})

	var body map[string]interface{}
	json.Unmarshal(w.Body.Bytes(), &body)
	arr := body["items"].([]interface{})
	if len(arr) != 2 {
		t.Errorf("got %d items, want 2", len(arr))
	}
	t.Logf("TopDigests OK: %d items", len(arr))
}

func TestHandler_Trend(t *testing.T) {
	mock := mockDataProxy(t, 0)
	defer mock.Close()
	svc := newTestService(mock.URL)

	w := httptest.NewRecorder()
	c, _ := gin.CreateTestContext(w)
	c.Request = httptest.NewRequest("GET", "/trend?start_time=2026-04-09T00:00:00Z&end_time=2026-04-09T03:00:00Z&metric=count", nil)

	startTime, endTime, ok := parseTimeRange(c)
	if !ok {
		t.Fatal("parseTimeRange failed")
	}
	metric := c.DefaultQuery("metric", "count")
	interval := trendInterval(startTime, endTime)
	sql := buildTrendSQL(startTime, endTime, metric, interval, nil)
	result, err := svc.queryDataProxy(c.Request.Context(), "test-cluster", sql, 60)
	if err != nil {
		t.Fatal(err)
	}
	points := make([]map[string]interface{}, 0)
	for _, row := range result.Rows {
		points = append(points, map[string]interface{}{"t": toFloat(rawAt(row, 0)), "v": toFloat(rawAt(row, 1))})
	}
	c.JSON(http.StatusOK, gin.H{"metric": metric, "interval_seconds": interval, "points": points})

	var body map[string]interface{}
	json.Unmarshal(w.Body.Bytes(), &body)
	if body["metric"] != "count" {
		t.Errorf("metric = %v", body["metric"])
	}
	pts := body["points"].([]interface{})
	if len(pts) != 2 {
		t.Errorf("got %d points, want 2", len(pts))
	}
	t.Logf("Trend OK: metric=%s, %d points, interval=%vs", body["metric"], len(pts), body["interval_seconds"])
}

func TestHandler_Detail(t *testing.T) {
	mock := mockDataProxy(t, 0)
	defer mock.Close()
	svc := newTestService(mock.URL)

	sql := buildDetailSQL(1775707648.457, "abc123", "conn_1")
	result, err := svc.queryDataProxy(context.Background(), "test-cluster", sql, 30)
	if err != nil {
		t.Fatal(err)
	}
	if len(result.Rows) == 0 {
		t.Fatal("no rows")
	}
	row := result.Rows[0]
	event := make(map[string]interface{})
	for i, col := range result.Columns {
		if i < len(row) {
			event[col] = row[i]
		}
	}
	event["cluster_id"] = "test-cluster"
	if event["digest"] != "abc123" {
		t.Errorf("digest = %v", event["digest"])
	}
	if event["query"] != "SELECT 1" {
		t.Errorf("query = %v", event["query"])
	}
	t.Logf("Detail OK: digest=%s, query=%s", event["digest"], event["query"])
}

func TestRawRecordPreservesDataProxyFields(t *testing.T) {
	resp := &DataProxyResponse{
		Columns: []string{"time", "query_time", "decoded_plan", "plan"},
		Rows: [][]interface{}{
			{1775707648.457, 1.5, `{"main":{"name":"TableReader_1"}}`, "raw plan text"},
		},
	}
	record := rawRecord(resp)
	if record == nil {
		t.Fatal("expected raw record")
	}
	if _, ok := record["cluster_id"]; ok {
		t.Error("raw record should not include transformed cluster_id")
	}
	if _, ok := record["query_time_ms"]; ok {
		t.Error("raw record should not include transformed query_time_ms")
	}
	if record["query_time"] != 1.5 {
		t.Errorf("query_time = %v, want raw seconds", record["query_time"])
	}
	if record["decoded_plan"] != `{"main":{"name":"TableReader_1"}}` {
		t.Errorf("decoded_plan = %v, want raw decoded_plan string", record["decoded_plan"])
	}
}

func TestBigQueryQuantileArrayConversion(t *testing.T) {
	percentiles := make([]interface{}, 101)
	for i := range percentiles {
		percentiles[i] = float64(i) / 100
	}

	if got := toMsFloatOrZero(quantileAt(percentiles, 50)); got != float64(500) {
		t.Fatalf("p50 ms = %v, want 500", got)
	}
	if got := toMsFloatOrZero(quantileAt(percentiles, 90)); got != float64(900) {
		t.Fatalf("p90 ms = %v, want 900", got)
	}
	if got := toMsFloatOrZero(quantileAt(percentiles, 99)); got != float64(990) {
		t.Fatalf("p99 ms = %v, want 990", got)
	}
	if got := quantileAt(percentiles, 101); got != nil {
		t.Fatalf("out-of-range quantile = %v, want nil", got)
	}
}

// ---- Data Proxy client ----

func TestQueryDataProxy_Success(t *testing.T) {
	mock := mockDataProxy(t, 5)
	defer mock.Close()
	svc := newTestService(mock.URL)

	result, err := svc.queryDataProxy(context.Background(), "c1", "SELECT 1", 10)
	if err != nil {
		t.Fatal(err)
	}
	if len(result.Columns) == 0 {
		t.Error("no columns")
	}
	if len(result.Rows) != 5 {
		t.Errorf("got %d rows, want 5", len(result.Rows))
	}
}

func TestQueryDataProxy_WithSynapseOptions(t *testing.T) {
	var captured DataProxyRequest
	mock := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if err := json.NewDecoder(r.Body).Decode(&captured); err != nil {
			t.Fatal(err)
		}
		json.NewEncoder(w).Encode(DataProxyResponse{
			Columns: []string{"ok"},
			Rows:    [][]interface{}{{true}},
		})
	}))
	defer mock.Close()
	svc := newTestService(mock.URL)

	_, err := svc.queryDataProxy(context.Background(), "c1", "SELECT TOP (1) * FROM clinic_data_proxy.slow_query_logs", 10, slowlogDataProxyOptionsForDialect(slowlogQueryDialectSynapse))
	if err != nil {
		t.Fatal(err)
	}
	if captured.Options.Vendor != "azure" {
		t.Fatalf("vendor = %q, want azure", captured.Options.Vendor)
	}
	if captured.Options.SkipSQLRewrite {
		t.Fatal("skipSQLRewrite should be false for Synapse SQL")
	}
	if !captured.Options.SkipTableManagement {
		t.Fatal("skipTableManagement should be true for Synapse SQL")
	}
}

func TestQueryDataProxy_Error(t *testing.T) {
	mock := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		http.Error(w, "internal error", 500)
	}))
	defer mock.Close()
	svc := newTestService(mock.URL)

	_, err := svc.queryDataProxy(context.Background(), "c1", "SELECT 1", 5)
	if err == nil {
		t.Fatal("expected error")
	}
	if !strings.Contains(err.Error(), "500") {
		t.Errorf("error should contain 500: %v", err)
	}
}