package service

import (
	"archive/zip"
	"context"
	"dataservices/commons/config"
	chttp "dataservices/commons/http"
	"dataservices/commons/logger"
	"dataservices/commons/redisutil"
	"dataservices/commons/utils"
	dto "dataservices/internal/dto/dataapi_control"
	"encoding/base64"
	"log"

	"fmt"
	"io"
	"os"
	"path/filepath"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/avast/retry-go"
	"github.com/go-redis/redis/v8"
	"github.com/golang/snappy"
	pb "github.com/tidbcloud/dataservices/idl/pbgen/proto/regioncentral"
	tierr "github.com/tidbcloud/tiinf/errs"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/protobuf/proto"

	"net/http"
	"net/url"

	awsconfig "github.com/aws/aws-sdk-go-v2/config"
	manager "github.com/aws/aws-sdk-go-v2/feature/s3/manager"
	s3 "github.com/aws/aws-sdk-go-v2/service/s3"
	"github.com/aws/aws-sdk-go/aws"
	jsoniter "github.com/json-iterator/go"
)

var emptyChunk pb.DataAppLogCacheChunk

func uint64ToStr(n uint64) string {
	return strconv.FormatUint(n, 10)
}

func getChunkKey(queryId string, chunkId uint64) string {
	return fmt.Sprintf("dataapp_log_cache:%s:%d", queryId, chunkId)
}

func getQueryInfoKey(queryId string, tenantId string) string {
	return fmt.Sprintf("dataapp_log_query_Info:%s:%s", tenantId, queryId)
}

func getQueryIdGeneratorKey() string {
	return "dataapp_log_queryId_gen"
}

func min(a, b int) int {
	if a < b {
		return a
	}
	return b
}

func max(a, b int) int {
	if a > b {
		return a
	}
	return b
}

type dataappLogSvc struct {
	cnf             config.CustomLogQueryConfig
	httpClient      *http.Client
	cacheMaster     *redis.Client
	S3PresignClient *s3.PresignClient
	S3Client        *s3.Client
}

var globalClient dataappLogSvc

func InitDataAppLogService(cnf *config.CustomLogQueryConfig) error {
	globalClient.cnf = *cnf
	globalClient.httpClient = &http.Client{Timeout: time.Duration(cnf.QueryLokiTimeoutSec) * time.Second}
	globalClient.cacheMaster = redisutil.LogQueryRedisClient()
	s3cfg, err := awsconfig.LoadDefaultConfig(context.Background())
	if err != nil {
		return err
	}
	globalClient.S3Client = s3.NewFromConfig(s3cfg)
	globalClient.S3PresignClient = s3.NewPresignClient(globalClient.S3Client)
	return nil
}

func (c *dataappLogSvc) queryLoki(ctx context.Context, param *dto.QueryUserExecutionLogParam, limit uint64, timeRev bool) ([]*pb.DataAppLogLine, error) {

	queryUrl := url.URL{
		Scheme: "http",
		Host:   fmt.Sprintf("%s:%d", c.cnf.LokiAddr, c.cnf.LokiPort),
		Path:   "/loki/api/v1/query_range",
	}
	query := queryUrl.Query()
	query.Add("limit", uint64ToStr(limit))
	query.Add("start", uint64ToStr(param.Start))
	query.Add("end", uint64ToStr(param.End))
	if !timeRev {
		query.Add("direction", "forward")
	}
	stream := make(map[string]string)
	if param.Source == "elt-worker" {
		if len(param.DataApp) > 0 {
			stream["dataapp"] = param.DataApp
		}
		if len(param.ClusterID) > 0 {
			stream["cluster"] = param.ClusterID
		}
	} else {
		stream["dataapp"] = param.DataApp
		stream["cluster"] = param.ClusterID
	}
	if len(param.LogLevel) > 0 {
		stream["log_level"] = param.LogLevel
	}

	streamStr := []string{}
	for k, v := range stream {
		streamStr = append(streamStr, fmt.Sprintf("%s=\"%s\"", k, v))
	}
	if len(streamStr) == 0 {
		streamStr = append(streamStr, `dataapp=~"[0-9]+"`)
	}

	filter := make([]string, 0, 10)
	if len(param.TraceID) > 0 {
		filter = append(filter, fmt.Sprintf(`|="trace_id\":\"%s\""`, param.TraceID))
	}
	if len(param.ReqID) > 0 {
		filter = append(filter, fmt.Sprintf(`|="req_id\":\"%s\""`, param.ReqID))
	}
	if len(param.EndpointID) > 0 {
		filter = append(filter, fmt.Sprintf(`|="endpoint_id\":\"%s\""`, param.EndpointID))
	}
	logQl := fmt.Sprintf(`{%s}%s`, strings.Join(streamStr, ","), strings.Join(filter, " "))
	query.Add("query", logQl)
	queryUrl.RawQuery = query.Encode()
	req, err := http.NewRequest(http.MethodGet, queryUrl.String(), nil)
	if err != nil {
		return nil, tierr.NewInternalServerError("req loki build req failed", err)
	}
	req.Header.Set("X-Scope-OrgID", param.Org)
	tilog.Info(ctx, "query url", zap.String("url", req.URL.String()), zap.String("org_id", param.Org), zap.String("region", config.Global().AwsConf.Region))

	var data []byte
	err = retry.Do(func() error {
		newctx, cancel := context.WithTimeout(ctx, 30*time.Second)
		defer cancel()
		respRaw, err := c.httpClient.Do(req.WithContext(newctx))
		if nil != err {
			return tierr.NewInternalServerError("req loki http req failed", err)
		}

		if respRaw.StatusCode != http.StatusOK {
			return tierr.NewInternalServerError(fmt.Sprintf("loki http code %d", respRaw.StatusCode), nil)
		}
		if !strings.Contains(respRaw.Header.Get("Content-Type"), "application/json") {
			return tierr.NewInternalServerError(fmt.Sprintf("loki content type not json: %s", respRaw.Header.Get("Content-Type")), nil)
		}
		defer respRaw.Body.Close()
		data, err = io.ReadAll(respRaw.Body)
		if err != nil {
			return tierr.NewInternalServerError(fmt.Sprintf("loki read resp body failed, %v", err), err)
		}
		return nil
	}, retry.Attempts(10),
		retry.Delay(100*time.Millisecond),
		retry.DelayType(retry.BackOffDelay),
		retry.MaxDelay(3*time.Second),
		retry.LastErrorOnly(true),
		retry.OnRetry(func(n uint, err error) {
			logger.Error(ctx, "query loki failed", zap.Uint("times", n), zap.String("org_id", param.Org), zap.String("region", config.Global().AwsConf.Region), zap.Error(err))
		}),
	)
	if nil != err {
		return nil, err
	}

	var resp dto.LokiQueryRes
	if err := jsoniter.Unmarshal(data, &resp); err != nil {
		return nil, tierr.NewInternalServerError(fmt.Sprintf("unmarshal json failed, %v", err), err)
	}
	if resp.Status != "success" {
		return nil, tierr.NewInternalServerError(fmt.Sprintf("loki query not success, status is %s", resp.Status), nil)
	}
	realLog := make([]*pb.DataAppLogLine, 0, len(resp.Data.Result))
	for _, stream := range resp.Data.Result {
		for _, item := range stream.Values {
			var logItem pb.DataAppLogLine
			logItem.LogTime, _ = strconv.ParseUint(item[0], 10, 64)

			if err := jsoniter.UnmarshalFromString(item[1], &logItem); err != nil {
				tilog.Error(ctx, "unmarshal failed", zap.String("raw", item[1]), zap.String("err", err.Error()))
				continue
			}
			logItem.DataappId = stream.Stream["dataapp"]
			realLog = append(realLog, &logItem)
		}
	}
	sort.Slice(realLog, func(i, j int) bool {
		return realLog[i].LogTime > realLog[j].LogTime || (realLog[i].LogTime == realLog[j].LogTime && realLog[i].TraceId > realLog[j].TraceId)
	})
	return realLog, nil
}

// TODO: review
func (c *dataappLogSvc) splitStreamToChunks(stream []*pb.DataAppLogLine, pageForward bool) []*pb.DataAppLogCacheChunk {
	if len(stream) == 0 {
		return []*pb.DataAppLogCacheChunk{}
	}
	chunksNum := (len(stream) + int(c.cnf.CacheChunkSize) - 1) / int(c.cnf.CacheChunkSize)
	chunks := make([]*pb.DataAppLogCacheChunk, chunksNum)
	if pageForward {
		pos := 0
		for i := 0; i < len(chunks); i++ {
			chunks[i] = &pb.DataAppLogCacheChunk{
				Content: stream[pos:min(pos+int(c.cnf.CacheChunkSize), len(stream))],
			}
			pos += int(c.cnf.CacheChunkSize)
			if pos >= len(stream) {
				break
			}
		}
	} else {
		pos := len(stream)
		for i := len(chunks) - 1; i >= 0; i-- {
			chunks[i] = &pb.DataAppLogCacheChunk{
				Content: stream[max(0, pos-int(c.cnf.CacheChunkSize)):pos],
			}
			pos -= int(c.cnf.CacheChunkSize)
			if pos <= 0 {
				break
			}
		}
	}
	return chunks
}

func (c *dataappLogSvc) fetchChunksFromLokiWithSentinel(ctx context.Context, query *dto.QueryUserExecutionLogParam, chunksNum uint64, pageForward bool, redunantQuery uint64, sentinel *pb.DataAppLogCacheLineFingerPrint) ([]*pb.DataAppLogCacheChunk, error) {
	// logs stream sort by ts desc :
	// later -------------------------> earlier
	//                |       |
	// qrange:       end    start
	// start<=result log ts <end
	start := query.Start
	end := query.End
	if pageForward { //page forward to get earlier log, move `end` to the sentinel
		end = sentinel.Ts + 1
	} else { // page back to get later log, move `start` to the sentinel
		start = sentinel.Ts
	}
	size := c.cnf.CacheChunkSize*chunksNum + redunantQuery
	// dont change origin query
	newQuery := *query
	newQuery.Start = start
	newQuery.End = end
	stream, err := c.queryLoki(ctx, &newQuery, size, pageForward)
	if err != nil {
		return nil, err
	}
	eof := len(stream) < int(size)
	// deduplication
	if pageForward {
		truncPos := 0
		for i := range stream {
			if stream[i].LogTime == sentinel.Ts && stream[i].TraceId == sentinel.TraceId {
				truncPos = i + 1
				break
			}
		}
		stream = stream[truncPos:]
		log.Printf("found trunc pos %d, left %d\n", truncPos, len(stream))
		if len(stream) > int(c.cnf.CacheChunkSize*chunksNum) {
			stream = stream[:c.cnf.CacheChunkSize*chunksNum]
		}
	} else { //backward
		truncPos := len(stream)
		for i := len(stream) - 1; i >= 0; i-- {
			if stream[i].LogTime == sentinel.Ts && stream[i].TraceId == sentinel.TraceId {
				truncPos = i
				break
			}
		}
		stream = stream[:truncPos]
		if len(stream) > int(c.cnf.CacheChunkSize*chunksNum) {
			stream = stream[len(stream)-int(c.cnf.CacheChunkSize*chunksNum):]
		}
	}
	if uint64(len(stream)) < c.cnf.CacheChunkSize*chunksNum && !eof {
		if redunantQuery == 0 {
			redunantQuery = 10
		} else {
			redunantQuery *= 2
		}
		// not full, try again
		return c.fetchChunksFromLokiWithSentinel(ctx, query, chunksNum, pageForward, redunantQuery, sentinel)
	}
	return c.splitStreamToChunks(stream, pageForward), nil
}

func (c *dataappLogSvc) fetchChunkFromLokiFromBeginning(ctx context.Context, query *dto.QueryUserExecutionLogParam, chunksNum uint64) ([]*pb.DataAppLogCacheChunk, error) {
	log.Println("fetchChunkFromLokiFromBeginning")
	size := c.cnf.CacheChunkSize * chunksNum
	stream, err := c.queryLoki(ctx, query, size, true)
	if err != nil {
		return nil, err
	}
	if len(stream) > int(c.cnf.CacheChunkSize*chunksNum) {
		stream = stream[:c.cnf.CacheChunkSize*chunksNum]
	}
	return c.splitStreamToChunks(stream, true), nil
}

func (c *dataappLogSvc) getChunkId(page, pageSize uint64) uint64 {
	return (page*pageSize + c.cnf.CacheChunkSize - 1) / c.cnf.CacheChunkSize
}

func (c *dataappLogSvc) decodePassback(passbackRaw string) (*pb.DataAppLogQueryPassback, error) {
	out, err := base64.RawURLEncoding.DecodeString(passbackRaw)
	if err != nil {
		return nil, err
	}
	var passback pb.DataAppLogQueryPassback
	if err := proto.Unmarshal(out, &passback); err != nil {
		return nil, err
	}
	return &passback, nil
}

func (c *dataappLogSvc) encodePassback(passback *pb.DataAppLogQueryPassback) string {
	raw, _ := proto.Marshal(passback)
	return base64.RawURLEncoding.EncodeToString(raw)
}

func (c *dataappLogSvc) writeChunksToCache(ctx context.Context, chunks []*pb.DataAppLogCacheChunk, startChunkId uint64, queryId string) error {
	pipeline := c.cacheMaster.Pipeline()
	for i, chunk := range chunks {
		data, _ := proto.Marshal(chunk)
		compressed := snappy.Encode(nil, data)
		pipeline.SetEX(ctx, getChunkKey(queryId, startChunkId+uint64(i)), compressed, time.Duration(c.cnf.CacheExpirySec)*time.Second)
	}
	_, err := pipeline.Exec(ctx)
	return err
}

func (c *dataappLogSvc) fillCacheForward(ctx context.Context, queryId string, query *dto.QueryUserExecutionLogParam, chunkNum int, hint *pb.DataAppLogQueryPassback) (*pb.DataAppLogCacheChunk, error) {
	var (
		lineFingerPrint pb.DataAppLogCacheLineFingerPrint
		chunks          []*pb.DataAppLogCacheChunk
		targetChunk     *pb.DataAppLogCacheChunk = &emptyChunk
		err             error
		fromBeginning   bool
		startChunkId    uint64
		fetched         int
	)
	log.Printf("fillCacheForward, trye to fetch %d chunks", chunkNum)
	if hint == nil {
		fromBeginning = true
		startChunkId = 1
	} else {
		if hint.ChunkLastLine == nil {
			return nil, tierr.NewInternalGeneralBusiness("fillCacheForward sentinel.lastLine==nil", nil)
		}
		startChunkId = hint.ChunkId + 1
		lineFingerPrint.TraceId = hint.ChunkLastLine.TraceId
		lineFingerPrint.Ts = hint.ChunkLastLine.Ts
	}
	for i := 0; i < chunkNum; i += int(c.cnf.MaxFetchChunkNum) {
		if fromBeginning && i == 0 {
			chunks, err = c.fetchChunkFromLokiFromBeginning(ctx, query, uint64(min(int(c.cnf.MaxFetchChunkNum), chunkNum-i)))
		} else {
			chunks, err = c.fetchChunksFromLokiWithSentinel(ctx, query, uint64(min(int(c.cnf.MaxFetchChunkNum), chunkNum-i)), true, 10, &lineFingerPrint)
		}
		if err != nil {
			tilog.Error(ctx, "fetch chunk failed", zap.Error(err))
			return nil, err
		}
		log.Printf("got chunks %d\n", len(chunks))
		if len(chunks) > 0 {
			chunkIdBegin := startChunkId + uint64(i)
			if err := c.writeChunksToCache(ctx, chunks, chunkIdBegin, queryId); err != nil {
				tilog.Error(ctx, "writeChunksToCache failed", zap.Error(err))
			}
			if len(chunks[len(chunks)-1].Content) > 0 {
				lineFingerPrint.Ts = chunks[len(chunks)-1].Content[len(chunks[len(chunks)-1].Content)-1].LogTime
				lineFingerPrint.TraceId = chunks[len(chunks)-1].Content[len(chunks[len(chunks)-1].Content)-1].TraceId
			}
			targetChunk = chunks[len(chunks)-1]
			fetched += len(chunks)
		} else {
			break
		}
	}
	if fetched < chunkNum {
		return &emptyChunk, nil
	}
	return targetChunk, nil
}

func (c *dataappLogSvc) fillCacheWithSentinelBackward(ctx context.Context, queryId string, query *dto.QueryUserExecutionLogParam, chunkNum int, hint *pb.DataAppLogQueryPassback) (*pb.DataAppLogCacheChunk, error) {
	if hint == nil || hint.ChunkFirstLine == nil {
		return nil, tierr.NewInternalGeneralBusiness("fillCacheWithSentinelBackward sentinel==nil", nil)
	}
	if chunkNum >= int(hint.ChunkId) {
		return nil, tierr.NewInternalGeneralBusiness("bad chunk num when page backforward", nil)
	}
	sentinel := hint.ChunkFirstLine
	var (
		lineFingerPrint pb.DataAppLogCacheLineFingerPrint
		chunks          []*pb.DataAppLogCacheChunk
		targetChunk     *pb.DataAppLogCacheChunk = &emptyChunk
		err             error
	)
	lineFingerPrint.TraceId = sentinel.TraceId
	lineFingerPrint.Ts = sentinel.Ts
	chunkIdBegin := int(hint.ChunkId)
	fetched := 0
	for i := chunkNum; i > 0; i -= int(c.cnf.MaxFetchChunkNum) {
		chunks, err = c.fetchChunksFromLokiWithSentinel(ctx, query, uint64(min(int(c.cnf.MaxFetchChunkNum), i)), false, 10, &lineFingerPrint)
		if err != nil {
			return nil, err
		}
		if len(chunks) > 0 {
			chunkIdBegin -= len(chunks)
			if err := c.writeChunksToCache(ctx, chunks, uint64(chunkIdBegin), queryId); err != nil {
				tilog.Error(ctx, "writeChunksToCache failed", zap.Error(err))
			}
			if len(chunks[0].Content) > 0 { // get the first line of the first chunk
				lineFingerPrint.Ts = chunks[0].Content[0].LogTime
				lineFingerPrint.TraceId = chunks[0].Content[0].TraceId
			}
			targetChunk = chunks[0]
			fetched += len(chunks)
		} else {
			break
		}
	}
	if fetched < chunkNum {
		return &emptyChunk, nil
	}
	return targetChunk, nil
}

func (c *dataappLogSvc) fillChacheOfSpecificChunk(ctx context.Context, chunkId uint64, queryId string, query *dto.QueryUserExecutionLogParam, redunantQuery uint64, hint *pb.DataAppLogQueryPassback) (*pb.DataAppLogCacheChunk, error) {
	if hint == nil || hint.ChunkFirstLine == nil {
		return nil, tierr.NewInternalGeneralBusiness("fillChacheOfSpecificChunk hint.firstline==nil", nil)
	}
	sentinel := hint.ChunkFirstLine
	start := query.Start
	end := sentinel.Ts + 1
	size := c.cnf.CacheChunkSize + redunantQuery
	// dont change origin query
	newQuery := *query
	newQuery.Start = start
	newQuery.End = end
	stream, err := c.queryLoki(ctx, &newQuery, size, true)
	if err != nil {
		return nil, err
	}
	eof := len(stream) < int(size)
	// find sentinel
	truncPos := 0
	for i := range stream {
		if stream[i].LogTime == sentinel.Ts && stream[i].TraceId == sentinel.TraceId {
			truncPos = i // do not +1, this line should be contained
			break
		}
	}
	log.Printf("sentinel %+v, trunc pos %d\n", sentinel, truncPos)
	stream = stream[truncPos:]
	if uint64(len(stream)) < c.cnf.CacheChunkSize && !eof {
		if redunantQuery == 0 {
			redunantQuery = 10
		} else {
			redunantQuery *= 2
		}
		// not full, try again
		return c.fillChacheOfSpecificChunk(ctx, chunkId, queryId, query, redunantQuery, hint)
	}
	if len(stream) == 0 {
		return &emptyChunk, nil
	}
	if len(stream) > int(c.cnf.CacheChunkSize) {
		stream = stream[:int(c.cnf.CacheChunkSize)]
	}
	chunk := &pb.DataAppLogCacheChunk{
		Content: stream,
	}
	if err := c.writeChunksToCache(ctx, []*pb.DataAppLogCacheChunk{chunk}, hint.ChunkId, queryId); err != nil {
		tilog.Error(ctx, "writeChunksToCache failed", zap.Error(err))
	}
	return chunk, nil
}

func (c *dataappLogSvc) fillCache(ctx context.Context, queryId string, expectedChunkId uint64, query *dto.QueryUserExecutionLogParam, passback *pb.DataAppLogQueryPassback) (*pb.DataAppLogCacheChunk, error) {
	if passback == nil {
		return c.fillCacheForward(ctx, queryId, query, int(expectedChunkId), nil)
	}
	sentinelId := passback.ChunkId
	if sentinelId < expectedChunkId { //page forwrad
		return c.fillCacheForward(ctx, queryId, query, int(expectedChunkId-sentinelId), passback)
	} else if sentinelId > expectedChunkId { //page backward
		if expectedChunkId < sentinelId-expectedChunkId { // from beginning
			return c.fillCacheForward(ctx, queryId, query, int(expectedChunkId), nil)
		}
		return c.fillCacheWithSentinelBackward(ctx, queryId, query, int(sentinelId-expectedChunkId), passback)
	}
	// equal
	return c.fillChacheOfSpecificChunk(ctx, expectedChunkId, queryId, query, 10, passback)
}

func (c *dataappLogSvc) tryFetchFromCache(ctx context.Context, queryId string, chunkId uint64) (*pb.DataAppLogCacheChunk, error) {
	// start from 1
	key := getChunkKey(queryId, chunkId)
	compressed, err := c.cacheMaster.Get(ctx, key).Bytes()
	if err != nil {
		return nil, err
	}
	raw, err := snappy.Decode(nil, compressed)
	if err != nil {
		return nil, err
	}
	var chunk pb.DataAppLogCacheChunk
	if err := proto.Unmarshal(raw, &chunk); err != nil {
		tilog.Warn(ctx, fmt.Sprintf("unmarshal chunk failed, %v", err), zap.String("chunk_key", key), zap.Error(err))
		return nil, redis.Nil
	}
	log.Printf("fetch from cach suc, qid %s, cid %d\n", queryId, chunkId)
	return &chunk, nil
}

func (c *dataappLogSvc) fetchChunk(ctx context.Context, queryId string, chunkId uint64, query *dto.QueryUserExecutionLogParam, passback *pb.DataAppLogQueryPassback) (*pb.DataAppLogCacheChunk, error) {
	log.Printf("fetchChunk chunkId %d\n", chunkId)
	chunk, err := c.tryFetchFromCache(ctx, queryId, chunkId)
	if err != nil && err != redis.Nil {
		return nil, err
	}
	if err == redis.Nil {
		chunk, err = c.fillCache(ctx, queryId, chunkId, query, passback)
	}
	if err != nil {
		return nil, err
	}
	log.Printf("chunkId %d fetched, chunk contesnt len %d", chunkId, len(chunk.Content))
	return chunk, nil
}

// 1. determine query_id
// 2. fetch from cache
// 2.1 fetched
// 2.1.1 if not the last page: return
// 2.1.2 if the last page in chunk:
// 2.1.2.1 keep result
// 2.1.2.2 page+=1, goto 1
// 2.2 not fetched: fetch from loki, set query start to the last line of this page, fetch size +=10, for the deduplicate redunancy
// 2.3 duplicate, write cache, return result

func ExecuteQuery(ctx context.Context, query *dto.QueryUserExecutionLogReq) (*dto.QueryUserExecutionLogRsp, error) {
	var c *dataappLogSvc = &globalClient
	var passback *pb.DataAppLogQueryPassback
	//step 1: confirm query info and gen/get query id
	var queryId string
	if len(query.Passback) > 0 {
		var err error
		passback, err = c.decodePassback(query.Passback)
		if err == nil {
			queryId = passback.QueryId
		} else {
			// invalid passback, clear them
			tilog.Error(ctx, "deocde passback failed", zap.Error(err))
			query.Passback = ""
			passback = nil
		}
	}
	if len(queryId) != 0 {
		equal, err := c.compareQueryInfo(ctx, queryId, query)
		if err != nil {
			return nil, err
		}
		if !equal {
			queryId = ""
		}
	}

	if len(queryId) == 0 {
		var err error
		queryId, err = c.generateQueryId(ctx)
		if err != nil {
			return nil, err
		}
		if err := c.storeQuery(ctx, queryId, query); err != nil {
			return nil, err
		}
		passback = nil
		query.Passback = ""
	}

	//step 2: fetch
	chunkId := c.getChunkId(query.Page, query.PageSize)

	chunk, err := c.fetchChunk(ctx, queryId, chunkId, &query.QueryUserExecutionLogParam, passback)
	if err != nil {
		tilog.Error(ctx, "fetchChunk failed", zap.Error(err))
		return nil, err
	}
	resp := dto.QueryUserExecutionLogRsp{}
	var passbackNew *pb.DataAppLogQueryPassback
	if len(chunk.Content) > 0 {
		log.Println(chunk.Content[0].Latency, chunk.Content[len(chunk.Content)-1].Latency)
		passbackNew = &pb.DataAppLogQueryPassback{
			QueryId: queryId,
			ChunkId: chunkId,
			ChunkLastLine: &pb.DataAppLogCacheLineFingerPrint{
				Ts:      chunk.Content[len(chunk.Content)-1].LogTime,
				TraceId: chunk.Content[len(chunk.Content)-1].TraceId,
			},
			ChunkFirstLine: &pb.DataAppLogCacheLineFingerPrint{
				Ts:      chunk.Content[0].LogTime,
				TraceId: chunk.Content[0].TraceId,
			},
		}
		resp.Passback = c.encodePassback(passbackNew)
	} else {
		resp.Passback = query.Passback
	}
	if (query.Page*query.PageSize)%c.cnf.CacheChunkSize == 0 && len(chunk.Content) == int(c.cnf.CacheChunkSize) {
		//chunk if full and we have got to the tail, so we need to check next chunk to decide HasMore
		newQuery := *query
		newQuery.Page += 1
		nextChunk, err := c.fetchChunk(ctx, queryId, chunkId+1, &newQuery.QueryUserExecutionLogParam, passbackNew)
		if err != nil {
			tilog.Error(ctx, "fetchChunk failed", zap.Error(err))
			return nil, err
		}
		if nextChunk != nil && len(nextChunk.Content) > 0 {
			resp.HasMore = true
		}
	}
	// got chunk
	offset := ((query.Page - 1) * query.PageSize) % c.cnf.CacheChunkSize
	if len(chunk.Content) <= int(offset) {
		resp.HasMore = false
		return &resp, nil
	}
	if int(offset+query.PageSize) < len(chunk.Content) {
		resp.HasMore = true
	}
	for _, logitem := range chunk.Content[offset:min(len(chunk.Content), int(offset+query.PageSize))] {
		resp.Data = append(resp.Data, convertLogPbToRaw(logitem))
	}

	return &resp, nil
}

func (c *dataappLogSvc) storeQuery(ctx context.Context, queryId string, query *dto.QueryUserExecutionLogReq) error {
	encoded, _ := jsoniter.Marshal(query.QueryUserExecutionLogParam)
	return c.cacheMaster.SetEX(ctx, getQueryInfoKey(queryId, query.Org), encoded, time.Duration(c.cnf.CacheExpirySec)*time.Second).Err()
}

func (c *dataappLogSvc) compareQueryInfo(ctx context.Context, queryId string, query *dto.QueryUserExecutionLogReq) (bool, error) {
	pip := c.cacheMaster.Pipeline()
	key := getQueryInfoKey(queryId, query.Org)
	res := pip.Get(ctx, key)
	pip.Expire(ctx, key, time.Duration(c.cnf.CacheExpirySec)*time.Second)
	_, err := pip.Exec(ctx)
	if err != nil {
		if err == redis.Nil {
			return false, nil
		}
		return false, err
	}
	storedStr, err := res.Result()
	if err != nil {
		if err == redis.Nil {
			err = nil
		}
		return false, err
	}
	var origin dto.QueryUserExecutionLogParam
	if err := jsoniter.UnmarshalFromString(storedStr, &origin); err != nil {
		return false, err
	}
	return origin == query.QueryUserExecutionLogParam, nil
}

func (c *dataappLogSvc) generateQueryId(ctx context.Context) (string, error) {
	id, err := c.cacheMaster.Incr(ctx, getQueryIdGeneratorKey()).Result()
	if err != nil {
		return "", err
	}
	return strconv.FormatInt(id, 10), nil
}

func convertLogPbToRaw(item *pb.DataAppLogLine) dto.CustomExecLogRaw {
	return dto.CustomExecLogRaw{
		LogTime:         item.LogTime,
		EndpointId:      item.EndpointId,
		EndpoitName:     item.EndpointName,
		Latency:         item.Latency,
		Level:           item.Level,
		Message:         item.Message,
		ReqID:           item.ReqId,
		TraceID:         item.TraceId,
		RspSize:         item.RspSize,
		QueryParam:      item.QueryParam,
		ReqHeader:       item.ReqHeader,
		ReqBody:         item.ReqBody,
		ReqUrl:          item.Url,
		ClientInfo:      item.Client,
		HttpCode:        int(item.HttpCode),
		HttpCodeMessage: item.HttpMsg,
		RspRowCnt:       int(item.RowCnt),
		RspCached:       item.Cache,
		Method:          item.Method,
		DataappID:       item.DataappId,
	}
}

func (c *dataappLogSvc) downloadedFileName(DataAppName string, t time.Time) string {
	return fmt.Sprintf("%s_logs_%s.zip", DataAppName, t.Format("20060102030405"))
}

func ExecuteDownload(ctx context.Context, query *dto.DownloadLogReq) error {
	var c *dataappLogSvc = &globalClient
	var passback *pb.DataAppLogQueryPassback
	//step 1: confirm query info and gen/get query id
	var queryId string
	if len(query.Passback) > 0 {
		var err error
		passback, err = c.decodePassback(query.Passback)
		if err == nil {
			queryId = passback.QueryId
		}
	}
	if len(queryId) != 0 {
		equal, err := c.compareQueryInfo(ctx, queryId, &query.QueryUserExecutionLogReq)
		if err != nil || !equal {
			queryId = ""
		}
	}
	if len(queryId) == 0 {
		var err error
		queryId, err = c.generateQueryId(ctx)
		if err != nil {
			return err
		}
	}
	utils.GO(func() {
		c.downloadAndWriteLogWorkRoutine(queryId, query.TaskId, &query.QueryUserExecutionLogParam, query.DataAppName)
	})
	return nil
}

func reportDownloadRes(taskId string, res dto.DownloadLogResult) {
	res.TaskId = taskId
	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()
	chttp.ReportDownloadDataAppLogResult(ctx, res)
}

func (c *dataappLogSvc) downloadAndWriteLogWorkRoutine(queryId string, taskId string, query *dto.QueryUserExecutionLogParam, appName string) {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second*time.Duration(c.cnf.DownloadTaskTimeoutSec))
	defer cancel()
	ts := time.Now().UTC()
	filename := c.downloadedFileName(appName, ts)
	if err := c.downloadAndWriteLog(ctx, queryId, filename, ts, query); err != nil {
		tilog.Error(ctx, "DownloadCustomLog failed, download to local failed", zap.String("taskid", taskId), zap.Error(err))
		reportDownloadRes(taskId, dto.DownloadLogResult{
			Result: "failed",
		})
		return
	}
	defer func() {
		os.Remove(filepath.Join(c.cnf.DownloadTempDir, filename))
	}()
	tilog.Info(ctx, "log fetch succ", zap.String("taskid", taskId))
	s3Url, err := c.uploadFile(ctx, filename)
	if err != nil {
		tilog.Error(ctx, "DownloadCustomLog failed, upload to s3 failed", zap.String("taskid", taskId), zap.Error(err))
		reportDownloadRes(taskId, dto.DownloadLogResult{
			Result: "failed",
		})
		return
	}
	tilog.Info(ctx, "log upload succ", zap.String("taskid", taskId))
	reportDownloadRes(taskId, dto.DownloadLogResult{
		Result: "success",
		URL:    s3Url,
	})
}

func (c *dataappLogSvc) downloadAndWriteLog(ctx context.Context, queryId string, zipfilename string, t time.Time, query *dto.QueryUserExecutionLogParam) error {
	output, err := os.Create(filepath.Join(c.cnf.DownloadTempDir, zipfilename))
	if err != nil {
		return err
	}
	defer output.Close()
	z := zip.NewWriter(output)
	w, err := z.Create(fmt.Sprintf("tcds_logs_%s.json", t.Format("20060102030405")))
	if err != nil {
		return err
	}
	_, err = w.Write([]byte(`{"logs":[`))
	if err != nil {
		return err
	}
	chunkId := uint64(1)
	begin := true
	var passback *pb.DataAppLogQueryPassback
	for fetched := 0; fetched < c.cnf.MaxDownloadLines; {
		log.Printf("downdload chunk %d\n", chunkId)
		chunk, err := c.fetchChunk(ctx, queryId, chunkId, query, passback)
		if err != nil {
			tilog.Error(ctx, "fetch chunk failed", zap.Error(err))
			return err
		}

		//write
		for _, item := range chunk.Content {
			if !begin {
				w.Write([]byte{','})
			} else {
				begin = false
			}
			data, _ := jsoniter.Marshal(convertLogPbToDownloadFormat(item))
			_, err = w.Write(data)
			if err != nil {
				return err
			}
			fetched++
			if fetched == c.cnf.MaxDownloadLines {
				break
			}
		}

		if len(chunk.Content) < int(c.cnf.CacheChunkSize) {
			//eof
			break
		}
		passback = &pb.DataAppLogQueryPassback{
			QueryId: queryId,
			ChunkId: chunkId,
			ChunkLastLine: &pb.DataAppLogCacheLineFingerPrint{
				Ts:      chunk.Content[len(chunk.Content)-1].LogTime,
				TraceId: chunk.Content[len(chunk.Content)-1].TraceId,
			},
			ChunkFirstLine: &pb.DataAppLogCacheLineFingerPrint{
				Ts:      chunk.Content[0].LogTime,
				TraceId: chunk.Content[0].TraceId,
			},
		}
		chunkId += 1
	}
	w.Write([]byte(`]}`))
	return z.Close()
}

func (c *dataappLogSvc) uploadFile(ctx context.Context, fileName string) (string, error) {
	file, err := os.Open(filepath.Join(c.cnf.DownloadTempDir, fileName))
	if err != nil {
		return "", err
	}
	defer file.Close()

	var partMiBs int64 = 10
	uploader := manager.NewUploader(c.S3Client, func(u *manager.Uploader) {
		u.PartSize = partMiBs * 1024 * 1024
	})
	_, err = uploader.Upload(ctx, &s3.PutObjectInput{
		Bucket: aws.String(c.cnf.S3Bucket),
		Key:    aws.String(fileName),
		Body:   file,
	})
	if err != nil {
		return "", err
	}

	res, err := c.S3PresignClient.PresignGetObject(ctx, &s3.GetObjectInput{
		Bucket: aws.String(c.cnf.S3Bucket),
		Key:    aws.String(fileName),
	}, func(opts *s3.PresignOptions) {
		opts.Expires = time.Duration(c.cnf.DownloadFileExp) * time.Second
	})
	if err != nil {
		return "", err
	}
	sign, _ := jsoniter.MarshalToString(res.SignedHeader)
	tilog.Info(ctx, "s3url", zap.String("method", res.Method), zap.String("url", res.URL), zap.String("header", sign))
	return res.URL, nil
}

func convertLogPbToDownloadFormat(item *pb.DataAppLogLine) *dto.CustomExecLogDownloadFormat {
	strToMap := func(s string) map[string]interface{} {
		res := make(map[string]interface{})
		jsoniter.UnmarshalFromString(s, &res)
		return res
	}
	res := dto.CustomExecLogDownloadFormat{
		DateTime:        time.Unix(int64(item.LogTime)/int64(time.Second), 0).UTC().Format("2006-01-02 03:04:05"),
		EndpoitName:     item.EndpointName,
		Latency:         item.Latency,
		Message:         strToMap(item.Message),
		ReqID:           item.ReqId,
		EndpointId:      item.EndpointId,
		RspSize:         item.RspSize,
		QueryParam:      strToMap(item.QueryParam),
		ReqHeader:       strToMap(item.ReqHeader),
		ReqBody:         item.ReqBody,
		ReqUrl:          item.Url,
		ClientInfo:      strToMap(item.Client),
		HttpCode:        int(item.HttpCode),
		HttpCodeMessage: item.HttpMsg,
		RspRowCnt:       int(item.RowCnt),
		RspCached:       item.Cache,
		Method:          item.Method,
	}
	return &res
}