package conprof

import (
	"archive/zip"
	"bytes"
	"context"
	_ "embed"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"os"
	"os/exec"
	"regexp"
	"strings"
	"time"

	gcpstorage "cloud.google.com/go/storage"
	"github.com/aws/aws-sdk-go/aws/awserr"
	"github.com/aws/aws-sdk-go/service/s3"
	"github.com/gin-gonic/gin"
	"go.uber.org/zap"

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

const (
	defaultTimeout = time.Second * 180
)

func shouldProxyConprofToO11y(deployType string) bool {
	switch strings.ToLower(deployType) {
	case "nextgen", o11y.PremiumType, o11y.NgHostType, o11y.ByocHostType:
		return true
	default:
		return false
	}
}

func shouldProxyConprofActionTokenToO11y(deployType string) bool {
	switch strings.ToLower(deployType) {
	case "nextgen", o11y.PremiumType, o11y.NgHostType:
		return true
	default:
		return false
	}
}

func InitConProfApi(g *gin.RouterGroup, s *Service) {
	g.GET("/continuous_profiling/group_profiles", s.handleList)
	g.GET("/continuous_profiling/group_profile/detail", s.handleDetail)
	g.GET("/continuous_profiling/action_token", s.handleActionToken)
	g.GET("/continuous_profiling/download", s.handleDownload)
	g.GET("/continuous_profiling/single_profile/view", s.handleSingleView)
}

type Service struct {
	ConProf  ConProf
	env      string
	cacheDir string
	o11yCli  *o11y.Client
}

func NewConProfService(env string, stores storage.StorageFactory, cacheDir string, o11yCli *o11y.Client) *Service {
	return &Service{
		ConProf:  NewConProf(env, stores),
		env:      env,
		cacheDir: cacheDir,
		o11yCli:  o11yCli,
	}
}

type errorResponse struct {
	Error    bool   `json:"error"`
	Message  string `json:"message"`
	Code     string `json:"code"`
	FullText string `json:"full_text"`
}

type clusterHeader struct {
	Provider   string `header:"x-provider" binding:"required"`
	Region     string `header:"x-region" binding:"required"`
	TenantID   string `header:"x-org-id" binding:"required"`
	ProjectID  string `header:"x-project-id"`
	ClusterID  string `header:"x-cluster-id" binding:"required"`
	DeployType string `header:"x-deploy-type"`
}

type listRequest struct {
	BeginTime int64 `json:"begin_time" form:"begin_time" binding:"required"`
	EndTime   int64 `json:"end_time" form:"end_time" binding:"required"`
}

type listResponseItem struct {
	Ts                  int64          `json:"ts"`
	ProfileDurationSecs int            `json:"profile_duration_secs"`
	State               string         `json:"state"`
	ComponentNum        map[string]int `json:"component_num"`
}

// @Summary		Get list of profile
// @Tags			conprof
// @Description	Get list of profile
// @Produce		json
// @Param			x-provider		header		string	true	"provider aws/gcp"
// @Param			x-region		header		string	true	"region"
// @Param			x-org-id		header		string	true	"organization id from dbaas"
// @Param			x-project-id	header		string	true	"project id from dbaas"
// @Param			x-cluster-id	header		string	true	"cluster id from dbaas"
// @Param			x-deploy-type	header		string	true	"deploy type: dedicated/shared/premium/nextgen/nextgen-host/byoc-host"
// @Param			begin_time		query		int		false	"begin time in unix seconds"
// @Param			end_time		query		int		true	"end time in unix seconds"
// @Success		200				{object}	[]listResponseItem
// @Router			/continuous_profiling/group_profiles [get]
func (s *Service) handleList(c *gin.Context) {
	f := recordHttpMetric(c, "/group_profiles")
	defer f()
	var header clusterHeader
	if err := c.ShouldBindHeader(&header); err != nil {
		c.JSON(http.StatusBadRequest, errorResponse{
			Error:    true,
			Message:  "invalid header",
			FullText: err.Error(),
		})
		return
	}
	var req listRequest
	if err := c.ShouldBindQuery(&req); err != nil {
		c.JSON(http.StatusBadRequest, errorResponse{
			Error:    true,
			Message:  "invalid params",
			FullText: err.Error(),
		})
		return
	}

	ctx, cancel := context.WithTimeout(c.Request.Context(), defaultTimeout)
	defer cancel()

	if shouldProxyConprofToO11y(header.DeployType) {
		err := utils.ProxyO11yByQueryCluster(ctx,
			c,
			s.o11yCli,
			header.DeployType,
			s.env,
			header.Provider,
			header.Region,
			header.TenantID,
			header.ProjectID,
			header.ClusterID,
			"/internal/diagnosis/api/v1/continuous_profiling/group_profiles")
		if err != nil {
			log.Error(ctx, "failed to proxy to o11y", zap.Error(err))
			c.JSON(http.StatusBadRequest, errorResponse{
				Error:    true,
				Message:  "failed to proxy to o11y",
				FullText: err.Error(),
			})
			return
		}
		return
	}

	metas, err := s.ConProf.List(c.Request.Context(), &ListParams{
		ClusterInfo: ClusterInfo(header),
		BeginTime:   req.BeginTime,
		EndTime:     req.EndTime,
	})
	if err != nil {
		if e, ok := err.(awserr.Error); ok && e.Code() == s3.ErrCodeNoSuchBucket {
			c.JSON(http.StatusBadRequest, errorResponse{
				Error:    true,
				Message:  fmt.Sprintf("aws region '%s' not deployed", header.Region),
				FullText: err.Error(),
			})
		} else if err == gcpstorage.ErrBucketNotExist {
			c.JSON(http.StatusBadRequest, errorResponse{
				Error:    true,
				Message:  fmt.Sprintf("gcp region '%s' not deployed", header.Region),
				FullText: err.Error(),
			})
		} else {
			c.JSON(http.StatusBadRequest, errorResponse{
				Error:    true,
				Message:  "failed to query conprof list",
				FullText: err.Error(),
			})
		}
		return
	}
	resp := convertProfileMetasToListResponseItems(metas)
	c.JSON(http.StatusOK, resp)
}

type detailRequest struct {
	Ts int64 `json:"ts" form:"ts" binding:"required"`
}

type detailResponse struct {
	Ts                  int64                   `json:"ts"`
	ProfileDurationSecs int                     `json:"profile_duration_secs"`
	State               string                  `json:"state"`
	TargetProfiles      []detailResponseProfile `json:"target_profiles"`
}

type detailResponseProfile struct {
	State       string                      `json:"state"`
	Error       string                      `json:"error"`
	ProfileType string                      `json:"profile_type"`
	Target      detailResponseProfileTarget `json:"target"`
}

type detailResponseProfileTarget struct {
	Address   string `json:"address"`
	Component string `json:"component"`
}

// @Summary		Get detail of profile
// @Tags			conprof
// @Description	Get detail of profile
// @Produce		json
// @Param			x-provider		header		string	true	"provider aws/gcp"
// @Param			x-region		header		string	true	"region"
// @Param			x-org-id		header		string	true	"organization id from dbaas"
// @Param			x-project-id	header		string	true	"project id from dbaas"
// @Param			x-cluster-id	header		string	true	"cluster id from dbaas"
// @Param			x-deploy-type	header		string	true	"deploy type: dedicated/shared/premium/nextgen/nextgen-host/byoc-host"
// @Param			ts				query		int		true	"timestamp in seconds"
// @Success		200				{object}	detailResponse
// @Router			/continuous_profiling/group_profile/detail [get]
func (s *Service) handleDetail(c *gin.Context) {
	f := recordHttpMetric(c, "/group_profile/detail")
	defer f()
	var header clusterHeader
	if err := c.ShouldBindHeader(&header); err != nil {
		c.JSON(http.StatusBadRequest, errorResponse{
			Error:    true,
			Message:  "invalid header",
			FullText: err.Error(),
		})
		return
	}
	var req detailRequest
	if err := c.ShouldBindQuery(&req); err != nil {
		c.JSON(http.StatusBadRequest, errorResponse{
			Error:    true,
			Message:  "invalid params",
			FullText: err.Error(),
		})
		return
	}

	ctx, cancel := context.WithTimeout(c.Request.Context(), defaultTimeout)
	defer cancel()

	if shouldProxyConprofToO11y(header.DeployType) {
		err := utils.ProxyO11yByQueryCluster(ctx,
			c,
			s.o11yCli,
			header.DeployType,
			s.env,
			header.Provider,
			header.Region,
			header.TenantID,
			header.ProjectID,
			header.ClusterID,
			"/internal/diagnosis/api/v1/continuous_profiling/group_profile/detail")
		if err != nil {
			log.Error(ctx, "failed to proxy to o11y", zap.Error(err))
			c.JSON(http.StatusBadRequest, errorResponse{
				Error:    true,
				Message:  "failed to proxy to o11y",
				FullText: err.Error(),
			})
			return
		}
		return
	}

	metas, err := s.ConProf.List(c.Request.Context(), &ListParams{
		ClusterInfo: ClusterInfo(header),
		BeginTime:   req.Ts,
		EndTime:     req.Ts,
	})
	if err != nil {
		if e, ok := err.(awserr.Error); ok && e.Code() == s3.ErrCodeNoSuchBucket {
			c.JSON(http.StatusBadRequest, errorResponse{
				Error:    true,
				Message:  fmt.Sprintf("aws region '%s' not deployed", header.Region),
				FullText: err.Error(),
			})
		} else if err == gcpstorage.ErrBucketNotExist {
			c.JSON(http.StatusBadRequest, errorResponse{
				Error:    true,
				Message:  fmt.Sprintf("gcp region '%s' not deployed", header.Region),
				FullText: err.Error(),
			})
		} else {
			c.JSON(http.StatusBadRequest, errorResponse{
				Error:    true,
				Message:  "failed to query conprof detail",
				FullText: err.Error(),
			})
		}
		return
	}
	resp := convertProfileMetasToDetailResponse(metas)
	c.JSON(http.StatusOK, resp)
}

// @Summary		Get action token for download or view profile
// @Tags			conprof
// @Description	Get action token for download or view profile
// @Produce		json
// @Param			x-provider		header		string	true	"provider aws/gcp"
// @Param			x-region		header		string	true	"region"
// @Param			x-org-id		header		string	true	"organization id from dbaas"
// @Param			x-project-id	header		string	true	"project id from dbaas"
// @Param			x-cluster-id	header		string	true	"cluster id from dbaas"
// @Param			x-deploy-type	header		string	true	"deploy type: dedicated/shared/premium/nextgen/nextgen-host/byoc-host"
// @Param			q				query		string	true	"target query string"
// @Success		200				{string}	string
// @Router			/continuous_profiling/action_token [get]
func (s *Service) handleActionToken(c *gin.Context) {
	f := recordHttpMetric(c, "/action_token")
	defer f()
	var header clusterHeader
	if err := c.ShouldBindHeader(&header); err != nil {
		c.JSON(http.StatusBadRequest, errorResponse{
			Error:    true,
			Message:  "invalid header",
			FullText: err.Error(),
		})
		return
	}

	ctx, cancel := context.WithTimeout(c.Request.Context(), defaultTimeout)
	defer cancel()

	if shouldProxyConprofActionTokenToO11y(header.DeployType) {
		err := utils.ProxyO11yByQueryCluster(ctx,
			c,
			s.o11yCli,
			header.DeployType,
			s.env,
			header.Provider,
			header.Region,
			header.TenantID,
			header.ProjectID,
			header.ClusterID,
			"/internal/diagnosis/api/v1/continuous_profiling/action_token")
		if err != nil {
			log.Error(ctx, "failed to proxy to o11y", zap.Error(err))
			c.JSON(http.StatusBadRequest, errorResponse{
				Error:    true,
				Message:  "failed to proxy to o11y",
				FullText: err.Error(),
			})
			return
		}
		return
	}

	q := c.Query("q")
	tokenDeployType := header.DeployType
	if header.DeployType == o11y.ByocHostType {
		log.Info(ctx, "o11y byoc host type, will use premium type for action token", zap.String("deployType", header.DeployType), zap.Any("clusterID", header.ClusterID))
		tokenDeployType = o11y.PremiumType
	}

	params := downloadTokenParams{
		Provider:         header.Provider,
		Region:           header.Region,
		TenantID:         header.TenantID,
		ProjectID:        header.ProjectID,
		ClusterID:        header.ClusterID,
		DeployType:       tokenDeployType,
		OriginDeployType: header.DeployType,
		Q:                q,
	}
	log.Info(ctx, "continuous_profiling action token params", zap.Any("params", params), zap.Any("ClusterID", header.ClusterID))
	b, _ := json.Marshal(params)
	token, err := utils.NewJWTString("conprof", string(b))
	if err != nil {
		c.JSON(http.StatusInternalServerError, errorResponse{
			Error:    true,
			Message:  "failed to generate jwt token",
			FullText: err.Error(),
		})
		return
	}
	log.Info(ctx, "continuous_profiling action token", zap.String("token", token))
	c.String(http.StatusOK, token)
}

type downloadTokenParams struct {
	Provider         string `json:"provider"`
	Region           string `json:"region"`
	TenantID         string `json:"tenant"`
	ProjectID        string `json:"project"`
	ClusterID        string `json:"cluster"`
	DeployType       string `json:"deploy_type"`
	OriginDeployType string `json:"origin_deploy_type"`
	Q                string `json:"q"`
}

//
//func (p *downloadTokenParams) UnmarshalJSON(data []byte) error {
//	type downloadTokenParamsAlias downloadTokenParams
//	if err := json.Unmarshal(data, (*downloadTokenParamsAlias)(p)); err != nil {
//		return err
//	}
//	if p.OriginDeployType != "" {
//		return nil
//	}
//
//	var legacy map[string]json.RawMessage
//	if err := json.Unmarshal(data, &legacy); err != nil {
//		return err
//	}
//	if raw, ok := legacy["or"+"gin_deploy_type"]; ok {
//		if err := json.Unmarshal(raw, &p.OriginDeployType); err != nil {
//			return err
//		}
//	}
//	return nil
//}

type downloadTokenQ struct {
	Ts          int64  `json:"ts" form:"ts" binding:"required"`                     // in secs
	Address     string `json:"address" form:"address" binding:"required"`           // e.g. 127.0.0.1:10080
	Component   string `json:"component" form:"component" binding:"required"`       // tidb/tikv/pd
	ProfileType string `json:"profile_type" form:"profile_type" binding:"required"` // cpu/heap/mutex/goroutine
	DataFormat  string `json:"data_format" form:"data_format"`                      // text/protobuf
}

func (q downloadTokenQ) toProfileMeta() ProfileMeta {
	return ProfileMeta{
		Ts:          q.Ts,
		Address:     q.Address,
		Component:   q.Component,
		ProfileType: q.ProfileType,
	}
}

// @Summary		Download profile
// @Tags			conprof
// @Description	Download profile
// @Param			token	query	string	true	"token for download or view profile"
// @Success		200		{file}	profile
// @Router			/continuous_profiling/download [get]
func (s *Service) handleDownload(c *gin.Context) {
	f := recordHttpMetric(c, "/download")
	defer f()

	ctx := c.Request.Context()
	token := c.Query("token")
	b, err := utils.ParseJWTString("conprof", token)
	if err != nil {
		c.JSON(http.StatusInternalServerError, errorResponse{
			Error:    true,
			Message:  "invalid jwt token",
			FullText: err.Error(),
		})
		return
	}
	var params downloadTokenParams
	if err := json.Unmarshal([]byte(b), &params); err != nil {
		log.Error(ctx, "invalid payload from jwt token", zap.Error(err))
		c.JSON(http.StatusInternalServerError, errorResponse{
			Error:    true,
			Message:  "invalid payload from jwt token",
			FullText: err.Error(),
		})
		return
	}
	log.Info(ctx, "continuous_profiling download params from jwt token", zap.Any("params", params))
	originDeployType := params.OriginDeployType
	if originDeployType == "" {
		log.Info(ctx, "origin deploy type is empty, use deploy type instead", zap.Any("deployType", params.DeployType))
		originDeployType = params.DeployType
	}
	if shouldProxyConprofToO11y(originDeployType) {
		ctx2, cancel := context.WithTimeout(ctx, defaultTimeout)
		defer cancel()
		err := utils.ProxyO11yByQueryCluster(ctx2,
			c,
			s.o11yCli,
			originDeployType,
			s.env,
			params.Provider,
			params.Region,
			params.TenantID,
			params.ProjectID,
			params.ClusterID,
			"/internal/diagnosis/api/v1/continuous_profiling/download")
		if err != nil {
			log.Error(ctx2, "failed to proxy to o11y", zap.Error(err))
			c.JSON(http.StatusBadRequest, errorResponse{
				Error:    true,
				Message:  "failed to proxy to o11y",
				FullText: err.Error(),
			})
			return
		}
		return
	}
	log.Info(ctx, "continuous_profiling download params", zap.Any("params", params))
	c.Request.URL.RawQuery = params.Q
	var q downloadTokenQ
	if err := c.ShouldBindQuery(&q); err != nil {
		c.JSON(http.StatusBadRequest, errorResponse{
			Error:    true,
			Message:  "invalid query params",
			FullText: err.Error(),
		})
		return
	}
	file, err := s.ConProf.Download(c.Request.Context(), &DownloadParams{
		ClusterInfo: ClusterInfo{
			Provider:  params.Provider,
			Region:    params.Region,
			TenantID:  params.TenantID,
			ProjectID: params.ProjectID,
			ClusterID: params.ClusterID,
		},
		ProfileMeta: q.toProfileMeta(),
	})
	if err != nil {
		log.Error(ctx, "continuous_profiling download failed", zap.Error(err))
		if e, ok := err.(awserr.Error); ok && e.Code() == s3.ErrCodeNoSuchBucket {
			c.JSON(http.StatusBadRequest, errorResponse{
				Error:    true,
				Message:  fmt.Sprintf("aws region '%s' not deployed", params.Region),
				FullText: err.Error(),
			})
		} else if err == gcpstorage.ErrBucketNotExist {
			c.JSON(http.StatusBadRequest, errorResponse{
				Error:    true,
				Message:  fmt.Sprintf("gcp region '%s' not deployed", params.Region),
				FullText: err.Error(),
			})
		} else {
			c.JSON(http.StatusBadRequest, errorResponse{
				Error:    true,
				Message:  "failed to download profile",
				FullText: err.Error(),
			})
		}
		return
	}
	filename := fmt.Sprintf("profile-%s.zip", time.Unix(q.Ts, 0).Format("2006-01-02_15-04-05"))
	log.Info(ctx, "continuous_profiling download filename", zap.String("filename", filename))
	c.Writer.Header().Set("Content-Disposition", "attachment; filename="+filename)
	c.Writer.WriteHeader(http.StatusOK)
	zw := zip.NewWriter(c.Writer)
	address := replaceNonAlphanumericWithDash(q.Address)
	filename = fmt.Sprintf("%v_%v_%v_%v.proto", q.ProfileType, q.Component, address, q.Ts)
	fw, err := zw.CreateHeader(&zip.FileHeader{
		Name:     filename,
		Method:   zip.Deflate,
		Modified: time.Now(),
	})
	if err != nil {
		log.Error(ctx, "failed to create zip header", zap.Error(err))
		c.JSON(http.StatusInternalServerError, errorResponse{
			Error:    true,
			Message:  "failed to create zip header",
			FullText: err.Error(),
		})
		return
	}
	_, err = fw.Write(file)
	if err != nil {
		c.JSON(http.StatusInternalServerError, errorResponse{
			Error:    true,
			Message:  "failed to write prof",
			FullText: err.Error(),
		})
		return
	}
	fw, err = zw.CreateHeader(&zip.FileHeader{
		Name:     "README.md",
		Method:   zip.Deflate,
		Modified: time.Now(),
	})
	if err != nil {
		c.JSON(http.StatusInternalServerError, errorResponse{
			Error:    true,
			Message:  "failed to create zip header for READNE.md",
			FullText: err.Error(),
		})
		return
	}
	_, err = fw.Write([]byte(downloadReadme))
	if err != nil {
		c.JSON(http.StatusInternalServerError, errorResponse{
			Error:    true,
			Message:  "failed to write README.md",
			FullText: err.Error(),
		})
		return
	}
	err = zw.Close()
	if err != nil {
		log.Error(ctx, "handle download request failed", zap.Error(err))
		return
	}
}

//go:embed jeprof.in
var jeprof string

// @Summary		Single view profile for speedscope
// @Tags			conprof
// @Description	Single view profile for speedscope
// @Param			token	query	string	true	"token for download or view profile"
// @Success		200		{file}	profile
// @Router			/continuous_profiling/single_profile/view [get]
func (s *Service) handleSingleView(c *gin.Context) {
	f := recordHttpMetric(c, "/single_profile/view")
	defer f()
	ctx := c.Request.Context()
	token := c.Query("token")
	log.Info(ctx, "continuous_profiling single view token", zap.String("token", token))
	b, err := utils.ParseJWTString("conprof", token)
	if err != nil {
		log.Error(ctx, "invalid jwt token", zap.Error(err))
		c.JSON(http.StatusInternalServerError, errorResponse{
			Error:    true,
			Message:  "invalid jwt token",
			FullText: err.Error(),
		})
		return
	}
	var params downloadTokenParams
	if err := json.Unmarshal([]byte(b), &params); err != nil {
		c.JSON(http.StatusInternalServerError, errorResponse{
			Error:    true,
			Message:  "invalid payload from jwt token",
			FullText: err.Error(),
		})
		log.Error(ctx, "invalid payload from jwt token", zap.Error(err))
		return
	}
	log.Info(ctx, "continuous_profiling download params from jwt token", zap.Any("params", params))
	originDeployType := params.OriginDeployType
	if originDeployType == "" {
		log.Info(ctx, "origin deploy type is empty, use deploy type instead", zap.Any("deployType", params.DeployType))
		originDeployType = params.DeployType
	}
	if shouldProxyConprofToO11y(originDeployType) {
		ctx2, cancel := context.WithTimeout(ctx, defaultTimeout)
		defer cancel()
		err := utils.ProxyO11yByQueryCluster(ctx2,
			c,
			s.o11yCli,
			originDeployType,
			s.env,
			params.Provider,
			params.Region,
			params.TenantID,
			params.ProjectID,
			params.ClusterID,
			"/internal/diagnosis/api/v1/continuous_profiling/single_profile/view")
		if err != nil {
			log.Error(ctx2, "failed to proxy to o11y", zap.Error(err))
			c.JSON(http.StatusBadRequest, errorResponse{
				Error:    true,
				Message:  "failed to proxy to o11y",
				FullText: err.Error(),
			})
			return
		}
		return
	}

	c.Request.URL.RawQuery = params.Q
	var q downloadTokenQ
	if err := c.ShouldBindQuery(&q); err != nil {
		c.JSON(http.StatusBadRequest, errorResponse{
			Error:    true,
			Message:  "invalid query params",
			FullText: err.Error(),
		})
		return
	}
	file, err := s.ConProf.Download(c.Request.Context(), &DownloadParams{
		ClusterInfo: ClusterInfo{
			Provider:  params.Provider,
			Region:    params.Region,
			TenantID:  params.TenantID,
			ProjectID: params.ProjectID,
			ClusterID: params.ClusterID,
		},
		ProfileMeta: q.toProfileMeta(),
	})
	if err != nil {
		if e, ok := err.(awserr.Error); ok && e.Code() == s3.ErrCodeNoSuchBucket {
			c.JSON(http.StatusBadRequest, errorResponse{
				Error:    true,
				Message:  fmt.Sprintf("aws region '%s' not deployed", params.Region),
				FullText: err.Error(),
			})
		} else if err == gcpstorage.ErrBucketNotExist {
			c.JSON(http.StatusBadRequest, errorResponse{
				Error:    true,
				Message:  fmt.Sprintf("gcp region '%s' not deployed", params.Region),
				FullText: err.Error(),
			})
		} else {
			c.JSON(http.StatusBadRequest, errorResponse{
				Error:    true,
				Message:  "failed to download profile",
				FullText: err.Error(),
			})
		}
		return
	}
	if shouldReturnCollapsedJeprofHeap(q, file) {
		textContent, err := convertJeprofHeapProfileToCollapsed(c.Request.Context(), file, s.cacheDir)
		if err != nil {
			c.JSON(http.StatusInternalServerError, errorResponse{
				Error:    true,
				Message:  "failed to convert heap profile",
				FullText: err.Error(),
			})
			return
		}
		file = textContent
		c.Writer.Header().Set("Content-Type", "text/plain")
	}
	c.Writer.WriteHeader(http.StatusOK)
	_, _ = c.Writer.Write(file)
}

func shouldReturnCollapsedJeprofHeap(q downloadTokenQ, profile []byte) bool {
	if !strings.EqualFold(q.ProfileType, "heap") || !strings.EqualFold(q.DataFormat, "text") {
		return false
	}
	switch strings.ToLower(q.Component) {
	case "tikv", "tiflash":
		return isJeprofHeapProfile(profile)
	default:
		return false
	}
}

func isJeprofHeapProfile(profile []byte) bool {
	profile = bytes.TrimSpace(profile)
	return bytes.HasPrefix(profile, []byte("--- symbol")) ||
		bytes.HasPrefix(profile, []byte("--- heap")) ||
		bytes.HasPrefix(profile, []byte("heap profile:")) ||
		bytes.HasPrefix(profile, []byte("heap_v2/"))
}

func isJeprofSymbolizedProfile(profile []byte) bool {
	profile = bytes.TrimSpace(profile)
	return bytes.HasPrefix(profile, []byte("--- symbol"))
}

func convertJeprofHeapProfileToCollapsed(ctx context.Context, profile []byte, cacheDir string) ([]byte, error) {
	if ctx == nil {
		ctx = context.Background()
	}
	profile = bytes.TrimSpace(profile)
	profileFile, err := os.CreateTemp(cacheDir, "jeprof-heap-*.prof")
	if err != nil {
		return nil, fmt.Errorf("failed to create heap profile temp file: %w", err)
	}
	profilePath := profileFile.Name()
	defer func() {
		if err := os.Remove(profilePath); err != nil {
			log.Warn(ctx, "failed to remove heap profile", zap.String("filepath", profilePath), zap.Error(err))
		}
	}()
	if _, err := profileFile.Write(profile); err != nil {
		_ = profileFile.Close()
		return nil, fmt.Errorf("failed to write heap profile: %w", err)
	}
	if err := profileFile.Close(); err != nil {
		return nil, fmt.Errorf("failed to close heap profile: %w", err)
	}

	args := []string{"/dev/stdin", "--collapsed"}
	if isJeprofSymbolizedProfile(profile) {
		args = append(args, profilePath)
	} else {
		binaryFile, err := os.CreateTemp(cacheDir, "jeprof-binary-*.bin")
		if err != nil {
			return nil, fmt.Errorf("failed to create heap profile binary placeholder: %w", err)
		}
		binaryPath := binaryFile.Name()
		if err := binaryFile.Close(); err != nil {
			_ = os.Remove(binaryPath)
			return nil, fmt.Errorf("failed to close heap profile binary placeholder: %w", err)
		}
		defer func() {
			if err := os.Remove(binaryPath); err != nil {
				log.Warn(ctx, "failed to remove heap profile binary placeholder", zap.String("filepath", binaryPath), zap.Error(err))
			}
		}()
		args = append(args, binaryPath, profilePath)
	}

	cmd := exec.CommandContext(ctx, "perl", args...) //nolint:gosec
	cmd.Stdin = strings.NewReader(jeprof)
	var stderr bytes.Buffer
	cmd.Stderr = &stderr
	textContent, err := cmd.Output()
	if err != nil {
		stderrText := strings.TrimSpace(stderr.String())
		if stderrText == "" {
			return nil, err
		}
		return nil, fmt.Errorf("%w: %s", err, stderrText)
	}
	if len(bytes.TrimSpace(textContent)) == 0 {
		stderrText := strings.TrimSpace(stderr.String())
		if stderrText == "" {
			return nil, errors.New("jeprof produced empty collapsed output")
		}
		return nil, fmt.Errorf("jeprof produced empty collapsed output: %s", stderrText)
	}
	return textContent, nil
}

func convertProfileMetasToListResponseItems(metas []ProfileMeta) []listResponseItem {
	resp := []listResponseItem{}
	if len(metas) == 0 {
		return resp
	}
	componentMap := map[int64]map[string]map[string]int{} // ts => component => address => count
	for _, meta := range metas {
		if _, ok := componentMap[meta.Ts]; !ok {
			componentMap[meta.Ts] = map[string]map[string]int{
				"pd":      {},
				"tidb":    {},
				"tikv":    {},
				"tiflash": {},
				"ticdc":   {},
			}
		}
		if _, ok := componentMap[meta.Ts][meta.Component]; !ok {
			// protect for unexpected component name
			componentMap[meta.Ts][meta.Component] = map[string]int{}
		}
		componentMap[meta.Ts][meta.Component][meta.Address]++
		if len(resp) == 0 || resp[len(resp)-1].Ts != meta.Ts {
			resp = append(resp, listResponseItem{
				Ts:                  meta.Ts,
				ProfileDurationSecs: 10,
				State:               "finished",
			})
		}
	}
	for n := range resp {
		resp[n].ComponentNum = map[string]int{}
		for component, m := range componentMap[resp[n].Ts] {
			resp[n].ComponentNum[component] = len(m)
		}
	}
	return resp
}

func convertProfileMetasToDetailResponse(metas []ProfileMeta) *detailResponse {
	if len(metas) == 0 {
		return nil
	}
	resp := &detailResponse{
		Ts:                  metas[0].Ts,
		ProfileDurationSecs: 10,
		State:               "finished",
	}
	for _, meta := range metas {
		resp.TargetProfiles = append(resp.TargetProfiles, detailResponseProfile{
			State:       "finished",
			ProfileType: meta.ProfileType,
			Target: detailResponseProfileTarget{
				Address:   meta.Address,
				Component: meta.Component,
			},
		})
	}
	return resp
}

var regexpInvalidFilenameChar = regexp.MustCompile(`[^a-zA-Z0-9]+`)

func replaceNonAlphanumericWithDash(s string) string {
	return regexpInvalidFilenameChar.ReplaceAllString(s, "-")
}

const downloadReadme = `
To review the profile data whose file name suffix is '.proto' interactively:

$ go tool pprof --http=127.0.0.1:6060 profile_xxx.proto
`