package handlers

import (
	"context"
	"errors"
	"fmt"
	"net/http"
	"net/http/httptest"
	"net/url"
	"strings"
	"testing"

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

	clinic_internal_api "github.com/pingcap-inc/clinic/clinic_data_service/pkg/clinic-internal-api"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/service"
)

const validBYOCStatementsDigest = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"

type fakeBYOCStatementsQuerier struct {
	result    *service.BYOCStatementsResult
	err       error
	calls     int
	orgID     uint64
	clusterID string
	operation service.BYOCStatementsOperation
	query     url.Values
}

func (f *fakeBYOCStatementsQuerier) Query(
	_ context.Context,
	orgID uint64,
	clusterID string,
	operation service.BYOCStatementsOperation,
	query url.Values,
) (*service.BYOCStatementsResult, error) {
	f.calls++
	f.orgID = orgID
	f.clusterID = clusterID
	f.operation = operation
	f.query = cloneBYOCStatementsValues(query)
	return f.result, f.err
}

func TestRegisterBYOCStatementsRoutes(t *testing.T) {
	gin.SetMode(gin.TestMode)
	router := gin.New()
	NewBYOCStatementsHandler(&fakeBYOCStatementsQuerier{}).RegisterRoutes(router)

	want := map[string]bool{
		http.MethodGet + " /data-proxy/api/v1/orgs/:orgId/clusters/:clusterId/statements":                  false,
		http.MethodGet + " /data-proxy/api/v1/orgs/:orgId/clusters/:clusterId/statements/available-fields": false,
		http.MethodGet + " /data-proxy/api/v1/orgs/:orgId/clusters/:clusterId/statements/statement-types":  false,
		http.MethodGet + " /data-proxy/api/v1/orgs/:orgId/clusters/:clusterId/statements/plans":            false,
		http.MethodGet + " /data-proxy/api/v1/orgs/:orgId/clusters/:clusterId/statements/plan-detail":      false,
	}
	for _, route := range router.Routes() {
		key := route.Method + " " + route.Path
		if _, ok := want[key]; ok {
			want[key] = true
		}
	}
	for route, found := range want {
		assert.Truef(t, found, "route %s was not registered", route)
	}
}

func TestBYOCStatementsFiveRoutesPreserveResponsesAndOperations(t *testing.T) {
	tests := []struct {
		name      string
		suffix    string
		body      string
		operation service.BYOCStatementsOperation
	}{
		{
			name:      "available fields",
			suffix:    "/available-fields",
			body:      `["digest","sum_latency"]`,
			operation: service.BYOCStatementsOperationAvailableFields,
		},
		{
			name:      "statement types",
			suffix:    "/statement-types",
			body:      `["Select","Update"]`,
			operation: service.BYOCStatementsOperationTypes,
		},
		{
			name:      "list",
			suffix:    "?begin_time=100&end_time=200",
			body:      `[{"digest":"` + validBYOCStatementsDigest + `"}]`,
			operation: service.BYOCStatementsOperationList,
		},
		{
			name:      "plans",
			suffix:    "/plans?begin_time=100&end_time=200&digest=" + validBYOCStatementsDigest,
			body:      `[{"plan_digest":"` + validBYOCStatementsDigest + `"}]`,
			operation: service.BYOCStatementsOperationPlans,
		},
		{
			name:      "plan detail",
			suffix:    "/plan-detail?begin_time=100&end_time=200&digest=" + validBYOCStatementsDigest,
			body:      `{"plan_digest":"` + validBYOCStatementsDigest + `"}`,
			operation: service.BYOCStatementsOperationPlanDetail,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			fake := successfulBYOCStatementsQuerier(tt.body)
			response := performBYOCStatementsRequest(t, fake, http.MethodGet, statementsPath("logical", tt.suffix), "", nil)

			require.Equal(t, http.StatusOK, response.Code, response.Body.String())
			assert.JSONEq(t, tt.body, response.Body.String())
			assert.Equal(t, 1, fake.calls)
			assert.Equal(t, uint64(123), fake.orgID)
			assert.Equal(t, "logical", fake.clusterID)
			assert.Equal(t, tt.operation, fake.operation)
		})
	}
}

func TestBYOCStatementsListForwardsRepeatedFiltersAndNormalizesStaticFields(t *testing.T) {
	fake := successfulBYOCStatementsQuerier(`[]`)
	query := url.Values{}
	query.Set("begin_time", "100")
	query.Set("end_time", "200")
	query.Add("schemas", "app")
	query.Add("schemas", "analytics")
	query.Add("stmt_types", "Select")
	query.Add("stmt_types", "Update")
	query.Set("text", "select account")
	query.Set("fields", " SUM_RU , digest_text ,exec_count ")

	response := performBYOCStatementsRequest(t, fake, http.MethodGet,
		statementsPath("logical", "?"+query.Encode()), "", nil)

	require.Equal(t, http.StatusOK, response.Code, response.Body.String())
	assert.Equal(t, 1, fake.calls)
	assert.Equal(t, service.BYOCStatementsOperationList, fake.operation)
	assert.Equal(t, []string{"app", "analytics"}, fake.query["schemas"])
	assert.Equal(t, []string{"Select", "Update"}, fake.query["stmt_types"])
	assert.Equal(t, "sum_ru,digest_text,exec_count", fake.query.Get("fields"))
	assert.Equal(t, "select account", fake.query.Get("text"))
}

func TestBYOCStatementsPlansAndPlanDetailAcceptCanonicalDigestForms(t *testing.T) {
	tests := []struct {
		name      string
		clusterID string
		suffix    string
		operation service.BYOCStatementsOperation
		wantPlans []string
	}{
		{
			name:      "plans accepts 64 hex digest",
			clusterID: "logical",
			suffix:    "/plans?begin_time=1&end_time=2&digest=" + validBYOCStatementsDigest,
			operation: service.BYOCStatementsOperationPlans,
		},
		{
			name:      "detail accepts omitted plans",
			clusterID: "logical",
			suffix:    "/plan-detail?begin_time=1&end_time=2&digest=" + validBYOCStatementsDigest,
			operation: service.BYOCStatementsOperationPlanDetail,
		},
		{
			name:      "detail accepts N A plan for physical cluster",
			clusterID: "physical-cluster",
			suffix:    "/plan-detail?begin_time=1&end_time=2&digest=" + validBYOCStatementsDigest + "&plans=N%2FA",
			operation: service.BYOCStatementsOperationPlanDetail,
			wantPlans: []string{"N/A"},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			fake := successfulBYOCStatementsQuerier(`{}`)
			if tt.operation == service.BYOCStatementsOperationPlans {
				fake.result.Body = []byte(`[]`)
			}
			response := performBYOCStatementsRequest(t, fake, http.MethodGet,
				statementsPath(tt.clusterID, tt.suffix), "", nil)

			require.Equal(t, http.StatusOK, response.Code, response.Body.String())
			assert.Equal(t, 1, fake.calls)
			assert.Equal(t, tt.clusterID, fake.clusterID)
			assert.Equal(t, tt.operation, fake.operation)
			assert.Equal(t, validBYOCStatementsDigest, fake.query.Get("digest"))
			assert.Equal(t, tt.wantPlans, fake.query["plans"])
		})
	}
}

func TestBYOCStatementsWindowBoundary(t *testing.T) {
	t.Run("259500 seconds accepted", func(t *testing.T) {
		fake := successfulBYOCStatementsQuerier(`[]`)
		response := performBYOCStatementsRequest(t, fake, http.MethodGet,
			statementsPath("logical", "?begin_time=0&end_time=259500"), "", nil)

		require.Equal(t, http.StatusOK, response.Code, response.Body.String())
		assert.Equal(t, 1, fake.calls)
	})

	t.Run("259501 seconds rejected", func(t *testing.T) {
		fake := successfulBYOCStatementsQuerier(`[]`)
		response := performBYOCStatementsRequest(t, fake, http.MethodGet,
			statementsPath("logical", "?begin_time=0&end_time=259501"), "", nil)

		assert.Equal(t, http.StatusBadRequest, response.Code, response.Body.String())
		assert.Zero(t, fake.calls)
	})
}

func TestBYOCStatementsRawQuerySizeBoundary(t *testing.T) {
	var raw strings.Builder
	raw.WriteString("begin_time=1&end_time=2")
	for i := 0; i < 63; i++ {
		raw.WriteString("&schemas=")
		raw.WriteString(strings.Repeat("a", 247))
	}
	raw.WriteString("&schemas=")
	raw.WriteString(strings.Repeat("b", 224))
	require.Equal(t, maxBYOCStatementsQueryBytes, raw.Len())

	t.Run("exactly 16 KiB accepted", func(t *testing.T) {
		fake := successfulBYOCStatementsQuerier(`[]`)
		response := performBYOCStatementsRequest(t, fake, http.MethodGet,
			statementsPath("logical", "?"+raw.String()), "", nil)

		require.Equal(t, http.StatusOK, response.Code, response.Body.String())
		assert.Equal(t, 1, fake.calls)
		assert.Len(t, fake.query["schemas"], 64)
	})

	t.Run("more than 16 KiB rejected", func(t *testing.T) {
		fake := successfulBYOCStatementsQuerier(`[]`)
		response := performBYOCStatementsRequest(t, fake, http.MethodGet,
			statementsPath("logical", "?"+raw.String()+"x"), "", nil)

		assert.Equal(t, http.StatusRequestEntityTooLarge, response.Code, response.Body.String())
		assert.Zero(t, fake.calls)
	})
}

func TestBYOCStatementsRejectsInvalidRequestsBeforeService(t *testing.T) {
	tooManySchemas := url.Values{"begin_time": {"1"}, "end_time": {"2"}}
	for i := 0; i <= maxBYOCStatementsCollectionValues; i++ {
		tooManySchemas.Add("schemas", fmt.Sprintf("schema-%d", i))
	}
	oversizedRawQuery := "begin_time=1&end_time=2&text=" + strings.Repeat("a", maxBYOCStatementsQueryBytes)

	tests := []struct {
		name       string
		method     string
		path       string
		body       string
		headers    http.Header
		wantStatus int
	}{
		{name: "available fields query", method: http.MethodGet, path: statementsPath("logical", "/available-fields?usage=list"), wantStatus: http.StatusBadRequest},
		{name: "statement types query", method: http.MethodGet, path: statementsPath("logical", "/statement-types?limit=1"), wantStatus: http.StatusBadRequest},
		{name: "unknown", method: http.MethodGet, path: statementsPath("logical", "?begin_time=1&end_time=2&unknown=x"), wantStatus: http.StatusBadRequest},
		{name: "dataSource", method: http.MethodGet, path: statementsPath("logical", "?begin_time=1&end_time=2&dataSource=v2"), wantStatus: http.StatusBadRequest},
		{name: "data_source", method: http.MethodGet, path: statementsPath("logical", "?begin_time=1&end_time=2&data_source=v2"), wantStatus: http.StatusBadRequest},
		{name: "resource groups", method: http.MethodGet, path: statementsPath("logical", "?begin_time=1&end_time=2&resource_groups=rg"), wantStatus: http.StatusBadRequest},
		{name: "duplicate begin", method: http.MethodGet, path: statementsPath("logical", "?begin_time=1&begin_time=2&end_time=3"), wantStatus: http.StatusBadRequest},
		{name: "duplicate end", method: http.MethodGet, path: statementsPath("logical", "?begin_time=1&end_time=2&end_time=3"), wantStatus: http.StatusBadRequest},
		{name: "duplicate text", method: http.MethodGet, path: statementsPath("logical", "?begin_time=1&end_time=2&text=a&text=b"), wantStatus: http.StatusBadRequest},
		{name: "duplicate fields", method: http.MethodGet, path: statementsPath("logical", "?begin_time=1&end_time=2&fields=digest&fields=sum_ru"), wantStatus: http.StatusBadRequest},
		{name: "duplicate digest", method: http.MethodGet, path: statementsPath("logical", "/plans?begin_time=1&end_time=2&digest="+validBYOCStatementsDigest+"&digest="+validBYOCStatementsDigest), wantStatus: http.StatusBadRequest},
		{name: "duplicate schema name", method: http.MethodGet, path: statementsPath("logical", "/plan-detail?begin_time=1&end_time=2&digest="+validBYOCStatementsDigest+"&schema_name=a&schema_name=b"), wantStatus: http.StatusBadRequest},
		{name: "missing begin", method: http.MethodGet, path: statementsPath("logical", "?end_time=2"), wantStatus: http.StatusBadRequest},
		{name: "missing end", method: http.MethodGet, path: statementsPath("logical", "?begin_time=1"), wantStatus: http.StatusBadRequest},
		{name: "negative begin", method: http.MethodGet, path: statementsPath("logical", "?begin_time=-1&end_time=2"), wantStatus: http.StatusBadRequest},
		{name: "timestamp too large", method: http.MethodGet, path: statementsPath("logical", "?begin_time=1&end_time=253402300800"), wantStatus: http.StatusBadRequest},
		{name: "reverse window", method: http.MethodGet, path: statementsPath("logical", "?begin_time=2&end_time=1"), wantStatus: http.StatusBadRequest},
		{name: "invalid static field", method: http.MethodGet, path: statementsPath("logical", "?begin_time=1&end_time=2&fields=digest%2Cdrop_table"), wantStatus: http.StatusBadRequest},
		{name: "duplicate static field", method: http.MethodGet, path: statementsPath("logical", "?begin_time=1&end_time=2&fields=digest%2CDIGEST"), wantStatus: http.StatusBadRequest},
		{name: "invalid static statement type", method: http.MethodGet, path: statementsPath("logical", "?begin_time=1&end_time=2&stmt_types=SELECT"), wantStatus: http.StatusBadRequest},
		{name: "too many schemas", method: http.MethodGet, path: statementsPath("logical", "?"+tooManySchemas.Encode()), wantStatus: http.StatusBadRequest},
		{name: "short digest", method: http.MethodGet, path: statementsPath("logical", "/plans?begin_time=1&end_time=2&digest=abc"), wantStatus: http.StatusBadRequest},
		{name: "non hex digest", method: http.MethodGet, path: statementsPath("logical", "/plans?begin_time=1&end_time=2&digest="+strings.Repeat("z", 64)), wantStatus: http.StatusBadRequest},
		{name: "invalid plan digest", method: http.MethodGet, path: statementsPath("logical", "/plan-detail?begin_time=1&end_time=2&digest="+validBYOCStatementsDigest+"&plans=abc"), wantStatus: http.StatusBadRequest},
		{name: "malformed query", method: http.MethodGet, path: statementsPath("logical", "?begin_time=1&end_time=%ZZ"), wantStatus: http.StatusBadRequest},
		{name: "query exceeds 16 KiB", method: http.MethodGet, path: statementsPath("logical", "?"+oversizedRawQuery), wantStatus: http.StatusRequestEntityTooLarge},
		{name: "GET body", method: http.MethodGet, path: statementsPath("logical", "?begin_time=1&end_time=2"), body: "not allowed", wantStatus: http.StatusBadRequest},
		{name: "caller placement query", method: http.MethodGet, path: statementsPath("logical", "?begin_time=1&end_time=2&provider=aws"), wantStatus: http.StatusBadRequest},
		{name: "caller placement header", method: http.MethodGet, path: statementsPath("logical", "?begin_time=1&end_time=2"), headers: http.Header{"X-Provider": {"aws"}}, wantStatus: http.StatusBadRequest},
		{name: "invalid org", method: http.MethodGet, path: "/data-proxy/api/v1/orgs/not-a-number/clusters/logical/statements?begin_time=1&end_time=2", wantStatus: http.StatusBadRequest},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			fake := successfulBYOCStatementsQuerier(`[]`)
			response := performBYOCStatementsRequest(t, fake, tt.method, tt.path, tt.body, tt.headers)
			assert.Equal(t, tt.wantStatus, response.Code, response.Body.String())
			assert.Zero(t, fake.calls, "service must not run for invalid input")
			assert.Contains(t, response.Header().Get("Content-Type"), "application/json")
		})
	}
}

func TestBYOCStatementsRejectsEveryServerRoutingHeaderBeforeService(t *testing.T) {
	headers := []string{
		"X-Env",
		"X-Biz-Type",
		"X-Org-Id",
		"X-Region-Id",
		"X-Application-Id",
		"X-Deploy-Type",
		"X-Deploy-Type-V2",
		"X-O11y-Cluster-Id",
	}
	for _, header := range headers {
		t.Run(header, func(t *testing.T) {
			fake := successfulBYOCStatementsQuerier(`[]`)
			response := performBYOCStatementsRequest(t, fake, http.MethodGet,
				statementsPath("logical", "?begin_time=1&end_time=2"), "", http.Header{header: {"caller-controlled"}})

			assert.Equal(t, http.StatusBadRequest, response.Code, response.Body.String())
			assert.Zero(t, fake.calls)
			assert.Contains(t, response.Body.String(), "server controlled")
		})
	}
}

func TestBYOCStatementsMapsServiceErrorsWithoutLeakingDetails(t *testing.T) {
	tests := []struct {
		name       string
		err        error
		detail     bool
		wantStatus int
	}{
		{name: "canceled", err: context.Canceled, wantStatus: clientClosedStatus},
		{name: "deadline", err: context.DeadlineExceeded, wantStatus: http.StatusGatewayTimeout},
		{name: "unavailable", err: service.ErrBYOCStatementsUnavailable, wantStatus: http.StatusServiceUnavailable},
		{name: "invalid target", err: fmt.Errorf("%w: secret target", service.ErrInvalidBYOCStatementsTarget), wantStatus: http.StatusBadGateway},
		{name: "invalid response", err: fmt.Errorf("%w: secret SQL", service.ErrInvalidBYOCStatementsResponse), wantStatus: http.StatusBadGateway},
		{name: "target unsupported", err: &clinic_internal_api.UpstreamError{StatusCode: http.StatusBadRequest}, wantStatus: http.StatusBadRequest},
		{name: "target missing", err: &clinic_internal_api.UpstreamError{StatusCode: http.StatusNotFound}, wantStatus: http.StatusNotFound},
		{name: "target timeout", err: &clinic_internal_api.UpstreamError{StatusCode: http.StatusGatewayTimeout}, wantStatus: http.StatusGatewayTimeout},
		{name: "target rate limited", err: &clinic_internal_api.UpstreamError{StatusCode: http.StatusTooManyRequests}, wantStatus: http.StatusTooManyRequests},
		{name: "target unavailable", err: &clinic_internal_api.UpstreamError{StatusCode: http.StatusServiceUnavailable}, wantStatus: http.StatusServiceUnavailable},
		{name: "target internal", err: &clinic_internal_api.UpstreamError{StatusCode: http.StatusInternalServerError}, wantStatus: http.StatusBadGateway},
		{name: "invalid upstream request", err: &service.BYOCStatementsUpstreamError{StatusCode: http.StatusBadRequest}, wantStatus: http.StatusBadRequest},
		{name: "missing list data", err: &service.BYOCStatementsUpstreamError{StatusCode: http.StatusNotFound}, wantStatus: http.StatusNotFound},
		{name: "missing plan detail", err: &service.BYOCStatementsUpstreamError{StatusCode: http.StatusNotFound}, detail: true, wantStatus: http.StatusNotFound},
		{name: "upstream rate limited", err: &service.BYOCStatementsUpstreamError{StatusCode: http.StatusTooManyRequests}, wantStatus: http.StatusTooManyRequests},
		{name: "upstream timeout", err: &service.BYOCStatementsUpstreamError{StatusCode: http.StatusGatewayTimeout}, wantStatus: http.StatusGatewayTimeout},
		{name: "upstream unavailable", err: &service.BYOCStatementsUpstreamError{StatusCode: http.StatusServiceUnavailable}, wantStatus: http.StatusServiceUnavailable},
		{name: "upstream internal", err: &service.BYOCStatementsUpstreamError{StatusCode: http.StatusInternalServerError}, wantStatus: http.StatusBadGateway},
		{name: "unknown secret", err: errors.New("customer SQL SELECT super_secret_value"), wantStatus: http.StatusBadGateway},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			fake := &fakeBYOCStatementsQuerier{err: tt.err}
			path := statementsPath("logical", "?begin_time=1&end_time=2")
			if tt.detail {
				path = statementsPath("logical", "/plan-detail?begin_time=1&end_time=2&digest="+validBYOCStatementsDigest)
			}
			response := performBYOCStatementsRequest(t, fake, http.MethodGet, path, "", nil)

			assert.Equal(t, tt.wantStatus, response.Code, response.Body.String())
			assert.Equal(t, 1, fake.calls)
			assert.NotContains(t, response.Body.String(), "secret")
			assert.NotContains(t, response.Body.String(), "SELECT")
			assert.Contains(t, response.Header().Get("Content-Type"), "application/json")
		})
	}
}

func TestBYOCStatementsRejectsNilServiceResult(t *testing.T) {
	fake := &fakeBYOCStatementsQuerier{}
	response := performBYOCStatementsRequest(t, fake, http.MethodGet,
		statementsPath("logical", "?begin_time=1&end_time=2"), "", nil)

	assert.Equal(t, http.StatusBadGateway, response.Code, response.Body.String())
	assert.Equal(t, 1, fake.calls)
}

func successfulBYOCStatementsQuerier(body string) *fakeBYOCStatementsQuerier {
	return &fakeBYOCStatementsQuerier{result: &service.BYOCStatementsResult{
		Body:         []byte(body),
		Provider:     "aws",
		Region:       "us-west-2",
		DeployTypeV2: "byoc",
	}}
}

func statementsPath(clusterID, suffix string) string {
	return "/data-proxy/api/v1/orgs/123/clusters/" + clusterID + "/statements" + suffix
}

func performBYOCStatementsRequest(
	t *testing.T,
	fake *fakeBYOCStatementsQuerier,
	method string,
	path string,
	body string,
	headers http.Header,
) *httptest.ResponseRecorder {
	t.Helper()
	gin.SetMode(gin.TestMode)
	router := gin.New()
	NewBYOCStatementsHandler(fake).RegisterRoutes(router)
	request := httptest.NewRequest(method, path, strings.NewReader(body))
	for key, values := range headers {
		for _, value := range values {
			request.Header.Add(key, value)
		}
	}
	response := httptest.NewRecorder()
	router.ServeHTTP(response, request)
	return response
}