package activities

import (
	"context"
	"encoding/json"
	"errors"
	"net/http"
	"net/http/httptest"
	"testing"

	"github.com/gin-gonic/gin"
	pingcaplog "github.com/pingcap/log"
	"github.com/stretchr/testify/require"
	"go.opentelemetry.io/otel/trace"
	"go.uber.org/zap"
	"go.uber.org/zap/zapcore"
	"go.uber.org/zap/zaptest/observer"
	"gorm.io/gorm"

	"github.com/pingcap-inc/clinic/pkg/activityhub_client"
	"github.com/pingcap-inc/clinic/pkg/auth/session"
)

type fakeActivitySearcher struct {
	search func(context.Context, string, activityhub_client.SearchParams) (activityhub_client.ActivitiesResponse, error)
}

func (f fakeActivitySearcher) SearchActivities(ctx context.Context, targetID string, params activityhub_client.SearchParams) (activityhub_client.ActivitiesResponse, error) {
	return f.search(ctx, targetID, params)
}

func internalActivityRouter(service *EventManagerService) *gin.Engine {
	gin.SetMode(gin.TestMode)
	router := gin.New()
	group := router.Group("/clinic/api/v1")
	registerInternalRouter(group, service)
	return router
}

func TestQueryInternalActivitiesChecksDBAASOwnershipAndUsesQueryCursor(t *testing.T) {
	service := &EventManagerService{
		owns: func(clusterID string, orgID uint64) error {
			if clusterID != "cluster-1" || orgID != 1234567890123456789 {
				t.Fatalf("ownership check = (%q, %d)", clusterID, orgID)
			}
			return nil
		},
		searcher: fakeActivitySearcher{search: func(_ context.Context, targetID string, params activityhub_client.SearchParams) (activityhub_client.ActivitiesResponse, error) {
			if targetID != "cluster-1" {
				t.Fatalf("targetID = %q", targetID)
			}
			want := activityhub_client.SearchParams{
				BeginTS:  100,
				EndTS:    200,
				Catalog:  "diagnosis",
				Type:     "event",
				Name:     "slow-query",
				Severity: 2,
				Limit:    25,
				Cursor:   "query-cursor",
			}
			if params != want {
				t.Fatalf("params = %#v, want %#v", params, want)
			}
			return activityhub_client.ActivitiesResponse{
				Activities: []*activityhub_client.Activity{{Name: "slow-query"}},
				Partial:    true,
				Cursor:     "next-cursor",
			}, nil
		}},
	}
	router := internalActivityRouter(service)

	request := httptest.NewRequest(http.MethodGet,
		"/clinic/api/v1/internal/data-proxy/v1/orgs/1234567890123456789/clusters/cluster-1/activities?begin_ts=100&end_ts=200&catalog=diagnosis&type=event&name=slow-query&severity=2&limit=25&cursor=query-cursor",
		nil)
	request.Header.Set("cursor", "legacy-header-must-be-ignored")
	request.Header.Set("X-Trace-ID", "trace-1")
	response := httptest.NewRecorder()
	router.ServeHTTP(response, request)

	if response.Code != http.StatusOK {
		t.Fatalf("status = %d, body = %s", response.Code, response.Body.String())
	}
	var got activityhub_client.ActivitiesResponse
	if err := json.Unmarshal(response.Body.Bytes(), &got); err != nil {
		t.Fatal(err)
	}
	if len(got.Activities) != 1 || got.Activities[0].Name != "slow-query" || !got.Partial || got.Cursor != "next-cursor" {
		t.Fatalf("response = %#v", got)
	}
}

func TestQueryInternalActivitiesRejectsUnownedClusterBeforeSearch(t *testing.T) {
	service := &EventManagerService{
		owns: func(string, uint64) error { return gorm.ErrRecordNotFound },
		searcher: fakeActivitySearcher{search: func(context.Context, string, activityhub_client.SearchParams) (activityhub_client.ActivitiesResponse, error) {
			t.Fatal("search must not run for an unowned cluster")
			return activityhub_client.ActivitiesResponse{}, nil
		}},
	}
	router := internalActivityRouter(service)
	response := httptest.NewRecorder()
	router.ServeHTTP(response, httptest.NewRequest(http.MethodGet,
		"/clinic/api/v1/internal/data-proxy/v1/orgs/123/clusters/cluster-1/activities?begin_ts=1&end_ts=2", nil))

	if response.Code != http.StatusNotFound {
		t.Fatalf("status = %d, body = %s", response.Code, response.Body.String())
	}
	if got := response.Body.String(); got != `{"code":"NotFound","message":"cluster not found"}` {
		t.Fatalf("body = %s", got)
	}
}

func TestQueryInternalActivitiesValidatesParametersBeforeOwnership(t *testing.T) {
	service := &EventManagerService{
		owns: func(string, uint64) error {
			t.Fatal("ownership must not run for an invalid query")
			return nil
		},
	}
	router := internalActivityRouter(service)

	tests := []string{
		"/clinic/api/v1/internal/data-proxy/v1/orgs/not-a-number/clusters/cluster-1/activities?begin_ts=1&end_ts=2",
		"/clinic/api/v1/internal/data-proxy/v1/orgs/123/clusters/cluster-1/activities?end_ts=2",
		"/clinic/api/v1/internal/data-proxy/v1/orgs/123/clusters/cluster-1/activities?begin_ts=3&end_ts=2",
		"/clinic/api/v1/internal/data-proxy/v1/orgs/123/clusters/cluster-1/activities?begin_ts=1&end_ts=2&limit=19",
		"/clinic/api/v1/internal/data-proxy/v1/orgs/123/clusters/cluster-1/activities?begin_ts=1&end_ts=2&limit=101",
		"/clinic/api/v1/internal/data-proxy/v1/orgs/123/clusters/cluster-1/activities?begin_ts=1&end_ts=2&limit=1000",
		"/clinic/api/v1/internal/data-proxy/v1/orgs/123/clusters/cluster-1/activities?begin_ts=1&begin_ts=2&end_ts=3",
		"/clinic/api/v1/internal/data-proxy/v1/orgs/123/clusters/cluster-1/activities?begin_ts=1&end_ts=2&cursor=bad%0Avalue",
	}
	for _, path := range tests {
		response := httptest.NewRecorder()
		router.ServeHTTP(response, httptest.NewRequest(http.MethodGet, path, nil))
		if response.Code != http.StatusBadRequest {
			t.Errorf("%s: status = %d, body = %s", path, response.Code, response.Body.String())
		}
	}
}

func TestQueryInternalActivitiesAcceptsLimit100(t *testing.T) {
	called := false
	service := &EventManagerService{
		owns: func(string, uint64) error { return nil },
		searcher: fakeActivitySearcher{search: func(_ context.Context, _ string, params activityhub_client.SearchParams) (activityhub_client.ActivitiesResponse, error) {
			called = true
			require.Equal(t, 100, params.Limit)
			return activityhub_client.ActivitiesResponse{}, nil
		}},
	}
	response := httptest.NewRecorder()
	internalActivityRouter(service).ServeHTTP(response, httptest.NewRequest(http.MethodGet,
		"/clinic/api/v1/internal/data-proxy/v1/orgs/123/clusters/cluster-1/activities?begin_ts=1&end_ts=2&limit=100", nil))

	require.Equal(t, http.StatusOK, response.Code)
	require.True(t, called)
}

func TestLegacyQueryActivitiesPreservesExistingWireContract(t *testing.T) {
	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if r.URL.Path != "/search/cluster-legacy/from/0/to/2" {
			t.Errorf("path = %q", r.URL.Path)
		}
		if r.URL.Query().Get("limit") != "20" || r.URL.Query().Get("severity") != "99" {
			t.Errorf("query = %q", r.URL.RawQuery)
		}
		if r.Header.Get("x-activityhub-cursor") != "legacy-cursor" {
			t.Errorf("cursor = %q", r.Header.Get("x-activityhub-cursor"))
		}
		if r.Header.Get("x-activityhub-source") != "clinic" || r.Header.Get("x-activityhub-key") != "secret" {
			t.Errorf("service credentials were not forwarded to ActivityHub")
		}
		_, _ = w.Write([]byte(`{"status":"error","msg":"Activity not found.","result":null}`))
	}))
	defer upstream.Close()

	service := &EventManagerService{
		hubClient: activityhub_client.NewClient(upstream.URL, "clinic", "secret"),
		searcher: fakeActivitySearcher{search: func(_ context.Context, targetID string, params activityhub_client.SearchParams) (activityhub_client.ActivitiesResponse, error) {
			t.Fatal("legacy handler must not use the strict internal/Data Proxy search policy")
			return activityhub_client.ActivitiesResponse{}, nil
		}},
	}
	gin.SetMode(gin.TestMode)
	router := gin.New()
	router.Use(func(c *gin.Context) {
		c.Set(session.CtxUserKey, &session.CtxUser{Internal: true})
		c.Next()
	})
	group := router.Group("/clinic/api/v1")
	registerRouter(group, service)

	request := httptest.NewRequest(http.MethodGet,
		"/clinic/api/v1/activityhub/applications/123/targets/cluster-legacy/activities?begin_ts=not-a-number&end_ts=2&limit=1&severity=99", nil)
	request.Header.Set("cursor", "legacy-cursor")
	response := httptest.NewRecorder()
	router.ServeHTTP(response, request)
	if response.Code != http.StatusOK || response.Body.String() != `{"activities":null,"partial":false,"cursor":""}` {
		t.Fatalf("status = %d, body = %s", response.Code, response.Body.String())
	}
}

func TestClassifyActivitySearchError(t *testing.T) {
	t.Parallel()

	tests := []struct {
		err    error
		status int
		code   string
	}{
		{context.Canceled, clientClosedStatus, "Canceled"},
		{context.DeadlineExceeded, http.StatusGatewayTimeout, "DeadlineExceeded"},
		{activityhub_client.ErrUnavailable, http.StatusServiceUnavailable, "Unavailable"},
		{&activityhub_client.UpstreamError{StatusCode: http.StatusOK}, http.StatusBadGateway, "BadGateway"},
		{&activityhub_client.UpstreamError{StatusCode: http.StatusTooManyRequests}, http.StatusTooManyRequests, "ResourceExhausted"},
		{&activityhub_client.UpstreamError{StatusCode: http.StatusUnauthorized}, http.StatusServiceUnavailable, "Unavailable"},
		{&activityhub_client.UpstreamError{StatusCode: http.StatusNotFound}, http.StatusBadGateway, "BadGateway"},
		{errors.New("invalid JSON"), http.StatusBadGateway, "BadGateway"},
	}
	for _, test := range tests {
		status, code, _ := classifyActivitySearchError(test.err)
		if status != test.status || code != test.code {
			t.Errorf("classify(%v) = (%d, %q), want (%d, %q)", test.err, status, code, test.status, test.code)
		}
	}
}

func TestQueryInternalActivitiesLogsSafeUpstreamDiagnostics(t *testing.T) {
	core, observed := observer.New(zapcore.ErrorLevel)
	restoreLogger := pingcaplog.ReplaceGlobals(
		zap.New(core),
		&pingcaplog.ZapProperties{Core: core, Level: zap.NewAtomicLevelAt(zapcore.ErrorLevel)},
	)
	t.Cleanup(restoreLogger)

	service := &EventManagerService{
		owns: func(string, uint64) error { return nil },
		searcher: fakeActivitySearcher{search: func(context.Context, string, activityhub_client.SearchParams) (activityhub_client.ActivitiesResponse, error) {
			return activityhub_client.ActivitiesResponse{}, &activityhub_client.UpstreamError{
				StatusCode:    http.StatusNotFound,
				ProtocolError: false,
				RequestID:     "activityhub-request-1",
				AWSRequestID:  "aws-request-1",
				APIGatewayID:  "gateway-id-1",
			}
		}},
	}
	response := httptest.NewRecorder()
	request := httptest.NewRequest(http.MethodGet,
		"/clinic/api/v1/internal/data-proxy/v1/orgs/123/clusters/cluster-1/activities?begin_ts=1&end_ts=2", nil)
	request = request.WithContext(trace.ContextWithSpanContext(request.Context(), trace.NewSpanContext(trace.SpanContextConfig{
		TraceID: trace.TraceID{0x11, 0x11, 0x11, 0x11, 0x11, 0x11, 0x11, 0x11, 0x11, 0x11, 0x11, 0x11, 0x11, 0x11, 0x11, 0x11},
		SpanID:  trace.SpanID{1},
	})))
	internalActivityRouter(service).ServeHTTP(response, request)
	require.Equal(t, http.StatusBadGateway, response.Code)

	entries := observed.FilterMessage("query activities failed").All()
	require.Len(t, entries, 1)
	fields := entries[0].ContextMap()
	require.EqualValues(t, http.StatusBadGateway, fields["status"])
	require.EqualValues(t, http.StatusNotFound, fields["upstream_http_status"])
	require.Equal(t, false, fields["upstream_protocol_error"])
	require.Equal(t, "activityhub-request-1", fields["upstream_request_id"])
	require.Equal(t, "aws-request-1", fields["upstream_aws_request_id"])
	require.Equal(t, "gateway-id-1", fields["upstream_api_gateway_id"])
	require.Equal(t, "11111111111111111111111111111111", fields["trace_id"])
}