package conprof

import (
	"context"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"net/url"
	"os"
	"strconv"
	"strings"
	"testing"

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

	"github.com/pingcap-inc/clinic/pkg/ngm/utils"
)

type fakeConProf struct {
	download []byte
	params   *DownloadParams
}

func (f *fakeConProf) List(context.Context, *ListParams) ([]ProfileMeta, error) {
	return nil, nil
}

func (f *fakeConProf) Download(_ context.Context, params *DownloadParams) ([]byte, error) {
	f.params = params
	return f.download, nil
}

func Test_extractFilenameFromObjectKey(t *testing.T) {
	key := "0/1369847559691427383/1369847559691457760/1373933076662920573/profiles/" +
		"1698657360-pd-cpu-ZGItcGQtMC5kYi1wZC1wZWVyLnRpZGIxMzczOTMzMDc2NjYyOTIwNTczLnN2YzoyMzc5.log"
	r := extractFilenameFromObjectKey(key)
	require.Equal(t, "1698657360-pd-cpu-ZGItcGQtMC5kYi1wZC1wZWVyLnRpZGIxMzczOTMzMDc2NjYyOTIwNTczLnN2YzoyMzc5", r)

	key = "0/1369847559691427383/1369847559691457760/1373933076662920573/profiles/" +
		"1698657360-pd-cpu-ZGItcGQtMC5kYi1wZC1wZWVyLnRpZGIxMzczOTMzMDc2NjYyOTIwNTczLnN2YzoyMzc5.log.gz"
	r = extractFilenameFromObjectKey(key)
	require.Equal(t, "1698657360-pd-cpu-ZGItcGQtMC5kYi1wZC1wZWVyLnRpZGIxMzczOTMzMDc2NjYyOTIwNTczLnN2YzoyMzc5", r)
}

func Test_extractTsFromFilename(t *testing.T) {
	filename := "1698657360-pd-cpu-ZGItcGQtMC5kYi1wZC1wZWVyLnRpZGIxMzczOTMzMDc2NjYyOTIwNTczLnN2YzoyMzc5"
	require.Equal(t, int64(1698657360), extractTsFromFilename(filename))
}

func Test_extractProfileMetaFromFilename(t *testing.T) {
	filename := "1698657360-pd-cpu-ZGItcGQtMC5kYi1wZC1wZWVyLnRpZGIxMzczOTMzMDc2NjYyOTIwNTczLnN2YzoyMzc5"
	meta := extractProfileMetaFromFilename(filename)
	require.Equal(t, int64(1698657360), meta.Ts)
	require.Equal(t, "pd", meta.Component)
	require.Equal(t, "cpu", meta.ProfileType)
	require.Equal(t, "db-pd-0.db-pd-peer.tidb1373933076662920573.svc:2379", meta.Address)
}

func Test_longestCommonPrefix(t *testing.T) {
	require.Equal(t, "fa", longestCommonPrefix([]string{"fan", "fat", "fame"}))
	require.Equal(t, "", longestCommonPrefix([]string{"bat", "van", "cat"}))
}

func Test_shouldReturnCollapsedJeprofHeap(t *testing.T) {
	jeprofHeap := []byte("--- symbol\n---\n--- heap\nheap profile: 1: 10 [1: 10] @ heapprofile\n")
	rawHeapV2 := []byte("heap_v2/524288\n")
	tests := []struct {
		name    string
		q       downloadTokenQ
		profile []byte
		want    bool
	}{
		{
			name:    "tikv heap text",
			q:       downloadTokenQ{Component: "tikv", ProfileType: "heap", DataFormat: "text"},
			profile: jeprofHeap,
			want:    true,
		},
		{
			name:    "tiflash heap text",
			q:       downloadTokenQ{Component: "tiflash", ProfileType: "heap", DataFormat: "text"},
			profile: jeprofHeap,
			want:    true,
		},
		{
			name:    "tikv raw heap v2 text",
			q:       downloadTokenQ{Component: "tikv", ProfileType: "heap", DataFormat: "text"},
			profile: rawHeapV2,
			want:    true,
		},
		{
			name:    "tikv heap protobuf format",
			q:       downloadTokenQ{Component: "tikv", ProfileType: "heap", DataFormat: "protobuf"},
			profile: jeprofHeap,
			want:    false,
		},
		{
			name:    "tikv heap missing data format",
			q:       downloadTokenQ{Component: "tikv", ProfileType: "heap"},
			profile: jeprofHeap,
			want:    false,
		},
		{
			name:    "tikv heap text with protobuf bytes",
			q:       downloadTokenQ{Component: "tikv", ProfileType: "heap", DataFormat: "text"},
			profile: []byte{0, 0, 0, 0, 1, 2, 3},
			want:    false,
		},
		{
			name:    "tidb heap text",
			q:       downloadTokenQ{Component: "tidb", ProfileType: "heap", DataFormat: "text"},
			profile: jeprofHeap,
			want:    false,
		},
		{
			name:    "tikv cpu text",
			q:       downloadTokenQ{Component: "tikv", ProfileType: "cpu", DataFormat: "text"},
			profile: jeprofHeap,
			want:    false,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			require.Equal(t, tt.want, shouldReturnCollapsedJeprofHeap(tt.q, tt.profile))
		})
	}
}

func Test_isJeprofSymbolizedProfile(t *testing.T) {
	tests := []struct {
		name    string
		profile []byte
		want    bool
	}{
		{name: "symbolized", profile: []byte("--- symbol\nbinary=/bin/unknown\n"), want: true},
		{name: "raw heap v2", profile: []byte("heap_v2/524288\n"), want: false},
		{name: "raw heap profile", profile: []byte("heap profile: 1: 10 [1: 10] @ heapprofile\n"), want: false},
		{name: "protobuf", profile: []byte{0, 0, 0, 0, 1, 2, 3}, want: false},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			require.Equal(t, tt.want, isJeprofSymbolizedProfile(tt.profile))
		})
	}
}

func Test_convertJeprofHeapProfileToCollapsedWithSymbolizedHeap(t *testing.T) {
	profile := []byte(`--- symbol
binary=/bin/unknown
0x1 foo
---
--- heap
heap profile: 1: 10 [1: 10] @ heapprofile
1: 10 [1: 10] @ 0x1
MAPPED_LIBRARIES:
`)

	cacheDir := t.TempDir()
	collapsed, err := convertJeprofHeapProfileToCollapsed(context.Background(), profile, cacheDir)
	require.NoError(t, err)
	require.Equal(t, "foo 10\n", string(collapsed))
	requireEmptyDir(t, cacheDir)
}

func Test_convertJeprofHeapProfileToCollapsedWithRawHeap(t *testing.T) {
	tests := []struct {
		name    string
		profile []byte
	}{
		{
			name: "threaded heap v2",
			profile: []byte(`heap_v2/524288
t*: 1: 524288 [1: 524288]
@ 0x3 0x2
t*: 1: 524288 [1: 524288]
MAPPED_LIBRARIES:
`),
		},
		{
			name: "heap profile",
			profile: []byte(`heap profile: 1: 10 [1: 10] @ heapprofile
1: 10 [1: 10] @ 0x2
MAPPED_LIBRARIES:
`),
		},
		{
			name: "wrapped heap profile",
			profile: []byte(`--- heap
heap profile: 1: 10 [1: 10] @ heapprofile
1: 10 [1: 10] @ 0x2
MAPPED_LIBRARIES:
`),
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			cacheDir := t.TempDir()
			collapsed, err := convertJeprofHeapProfileToCollapsed(context.Background(), tt.profile, cacheDir)
			require.NoError(t, err)
			requireCollapsedStackText(t, collapsed)
			requireEmptyDir(t, cacheDir)
		})
	}
}

func Test_convertJeprofHeapProfileToCollapsedCleansUpAfterRawProfileError(t *testing.T) {
	cacheDir := t.TempDir()
	_, err := convertJeprofHeapProfileToCollapsed(context.Background(), []byte("heap_v2/not-a-rate\n"), cacheDir)
	require.Error(t, err)
	require.Contains(t, err.Error(), "Threaded heap profiles require v2 sampling with a sample rate")
	requireEmptyDir(t, cacheDir)
}

func Test_convertJeprofHeapProfileToCollapsedRejectsEmptyOutput(t *testing.T) {
	cacheDir := t.TempDir()
	profile := []byte("heap_v2/524288\nMAPPED_LIBRARIES:\n")
	_, err := convertJeprofHeapProfileToCollapsed(context.Background(), profile, cacheDir)
	require.Error(t, err)
	require.Contains(t, err.Error(), "empty collapsed output")
	requireEmptyDir(t, cacheDir)
}

func Test_handleSingleViewConvertsTiFlashHeapTextToCollapsedStacks(t *testing.T) {
	gin.SetMode(gin.TestMode)
	profile := []byte(`--- symbol
binary=/bin/unknown
0x1 foo
---
--- heap
heap profile: 1: 10 [1: 10] @ heapprofile
1: 10 [1: 10] @ 0x1
MAPPED_LIBRARIES:
`)
	conProf := &fakeConProf{download: profile}
	service := &Service{
		ConProf:  conProf,
		cacheDir: t.TempDir(),
	}
	q := url.Values{}
	q.Set("ts", "1")
	q.Set("address", "127.0.0.1:3930")
	q.Set("component", "tiflash")
	q.Set("profile_type", "heap")
	q.Set("data_format", "text")
	tokenPayload, err := json.Marshal(downloadTokenParams{
		Provider:  "aws",
		Region:    "us-west-2",
		TenantID:  "tenant",
		ProjectID: "project",
		ClusterID: "cluster",
		Q:         q.Encode(),
	})
	require.NoError(t, err)
	token, err := utils.NewJWTString("conprof", string(tokenPayload))
	require.NoError(t, err)

	w := httptest.NewRecorder()
	c, _ := gin.CreateTestContext(w)
	req, err := http.NewRequest(http.MethodGet, "/continuous_profiling/single_profile/view?token="+url.QueryEscape(token), nil)
	require.NoError(t, err)
	c.Request = req

	service.handleSingleView(c)

	require.Equal(t, http.StatusOK, w.Code)
	require.Equal(t, "foo 10\n", w.Body.String())
	require.Contains(t, w.Header().Get("Content-Type"), "text/plain")
	require.NotNil(t, conProf.params)
	require.Equal(t, "tiflash", conProf.params.Component)
	require.Equal(t, "heap", conProf.params.ProfileType)
}

func Test_handleSingleViewConvertsRawTiKVHeapV2ToCollapsedStacks(t *testing.T) {
	profile := []byte(`heap_v2/524288
t*: 1: 524288 [1: 524288]
@ 0x3 0x2
t*: 1: 524288 [1: 524288]
MAPPED_LIBRARIES:
`)
	cacheDir := t.TempDir()
	w, conProf := runSingleView(t, profile, cacheDir, "tikv")

	require.Equal(t, http.StatusOK, w.Code, w.Body.String())
	require.Contains(t, w.Header().Get("Content-Type"), "text/plain")
	requireCollapsedStackText(t, w.Body.Bytes())
	require.NotNil(t, conProf.params)
	require.Equal(t, "tikv", conProf.params.Component)
	require.Equal(t, "heap", conProf.params.ProfileType)
	requireEmptyDir(t, cacheDir)
}

func Test_handleSingleViewRejectsMalformedRawTiKVHeapV2(t *testing.T) {
	cacheDir := t.TempDir()
	w, _ := runSingleView(t, []byte("heap_v2/not-a-rate\n"), cacheDir, "tikv")

	require.Equal(t, http.StatusInternalServerError, w.Code)
	require.Contains(t, w.Body.String(), "failed to convert heap profile")
	require.Contains(t, w.Body.String(), "Threaded heap profiles require v2 sampling with a sample rate")
	require.NotContains(t, w.Header().Get("Content-Type"), "text/plain")
	requireEmptyDir(t, cacheDir)
}

func runSingleView(t *testing.T, profile []byte, cacheDir, component string) (*httptest.ResponseRecorder, *fakeConProf) {
	t.Helper()
	gin.SetMode(gin.TestMode)
	conProf := &fakeConProf{download: profile}
	service := &Service{
		ConProf:  conProf,
		cacheDir: cacheDir,
	}

	q := url.Values{}
	q.Set("ts", "1")
	q.Set("address", "127.0.0.1:3930")
	q.Set("component", component)
	q.Set("profile_type", "heap")
	q.Set("data_format", "text")
	tokenPayload, err := json.Marshal(downloadTokenParams{
		Provider:  "aws",
		Region:    "us-west-2",
		TenantID:  "tenant",
		ProjectID: "project",
		ClusterID: "cluster",
		Q:         q.Encode(),
	})
	require.NoError(t, err)
	token, err := utils.NewJWTString("conprof", string(tokenPayload))
	require.NoError(t, err)

	w := httptest.NewRecorder()
	c, _ := gin.CreateTestContext(w)
	req, err := http.NewRequest(http.MethodGet, "/continuous_profiling/single_profile/view?token="+url.QueryEscape(token)+"&output_type=text", nil)
	require.NoError(t, err)
	c.Request = req
	service.handleSingleView(c)
	return w, conProf
}

func requireCollapsedStackText(t *testing.T, collapsed []byte) {
	t.Helper()
	text := strings.TrimSpace(string(collapsed))
	require.NotEmpty(t, text)
	lines := strings.Split(text, "\n")
	for _, line := range lines {
		fields := strings.Fields(line)
		require.GreaterOrEqual(t, len(fields), 2, line)
		_, err := strconv.ParseFloat(fields[len(fields)-1], 64)
		require.NoError(t, err, line)
	}
}

func requireEmptyDir(t *testing.T, dir string) {
	t.Helper()
	entries, err := os.ReadDir(dir)
	require.NoError(t, err)
	require.Empty(t, entries)
}