package slowquery

import (
	"bufio"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"strconv"
	"strings"
	"sync"
	"time"

	gcs "cloud.google.com/go/storage"
	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
	"github.com/aws/aws-sdk-go-v2/service/s3/types"
	"github.com/pingcap/log"
	"go.uber.org/zap"

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

const hourlyTimeFormat = "2006010215"

type slowQueryV2 struct {
	stores storage.StorageFactory
}

func NewSlowQueryV2(stores storage.StorageFactory) SlowQuery {
	return &slowQueryV2{
		stores: stores,
	}
}

func (s *slowQueryV2) List(ctx context.Context, params *ListParams) ([]*SlowLog, error) {
	l := log.L().With(zap.String("provider", params.Provider),
		zap.String("region", params.Region),
		zap.String("tenant", params.TenantID),
		zap.String("project", params.ProjectID),
		zap.String("cluster", params.ClusterID),
		zap.Bool("sharedTier", params.SharedTier),
		zap.Int("beginTime", params.BeginTime),
		zap.Int("endTime", params.EndTime),
		zap.Int("limit", params.Limit),
		zap.String("orderBy", params.OrderBy),
		zap.String("text", params.Text),
		zap.String("digest", params.Digest),
		zap.String("db", fmt.Sprintf("%v", params.DB)),
		zap.String("plans", fmt.Sprintf("%v", params.Plans)),
		zap.Bool("desc", params.IsDesc))

	l.Info("slow query v2 list")
	defer func() {
		l.Info("slow query v2 list end")
	}()

	bucket := fmt.Sprintf("o11y-prod-shared-%s", params.Region)
	store, err := s.stores.Storage(ctx, params.Provider, params.Region)
	if err != nil {
		return nil, err
	}
	limit := params.Limit
	if limit == 0 {
		limit = defaultListLimit
	}
	beginHourTs := int64(params.BeginTime - (params.BeginTime % 3600))
	beginHour := time.Unix(beginHourTs, 0)
	begin := time.Unix(int64(params.BeginTime), 0)
	endHourTs := int64(params.EndTime - (params.EndTime % 3600))
	endHour := time.Unix(endHourTs, 0)
	end := time.Unix(int64(params.EndTime), 0)
	if params.EndTime%3600 != 0 {
		endHour = endHour.Add(time.Hour)
	}

	switch strings.ToLower(params.OrderBy) {
	case "", "timestamp":
		if params.IsDesc {
			return s.readSlowLogsOrderByTimeDesc(ctx, store, bucket, params, begin, end, beginHour, endHour, limit)
		}
		return s.readSlowLogsOrderByTimeAsc(ctx, store, bucket, params, begin, end, beginHour, endHour, limit)
	case "query_time", "memory_max", "total_keys", "process_keys":
		return s.readSlowLogsOrderBy(ctx, store, bucket, params, begin, end, beginHour, endHour, limit)
	default:
		return nil, ErrUnsupportedOrderBy
	}
}

func (s *slowQueryV2) readSlowLogsOrderByTimeAsc(
	ctx context.Context,
	store storage.Storage,
	bucket string,
	params *ListParams,
	begin, end time.Time,
	beginHour, endHour time.Time,
	limit int,
) ([]*SlowLog, error) {
	logs := []*SlowLog{}
	for cur := beginHour; cur.Before(endHour); cur = cur.Add(time.Hour) {
		hour := cur.UTC().Format(hourlyTimeFormat)
		err := s.readSlowLogsHour(ctx, store, bucket, params, begin, end, hour, func(sl *SlowLog) bool {
			logs = append(logs, sl)
			return len(logs) < limit
		})
		if err != nil {
			return nil, err
		}
		if len(logs) >= limit {
			break
		}
	}
	return logs, nil
}

func (s *slowQueryV2) readSlowLogsOrderByTimeDesc(
	ctx context.Context,
	store storage.Storage,
	bucket string,
	params *ListParams,
	begin, end time.Time,
	beginHour, endHour time.Time,
	limit int,
) ([]*SlowLog, error) {
	logs := []*SlowLog{}
	for cur := endHour.Add(-time.Hour); cur.After(beginHour) || cur.Equal(beginHour); cur = cur.Add(-time.Hour) {
		hour := cur.UTC().Format(hourlyTimeFormat)
		hourlyLogs := []*SlowLog{}
		err := s.readSlowLogsHour(ctx, store, bucket, params, begin, end, hour, func(sl *SlowLog) bool {
			hourlyLogs = append(hourlyLogs, sl)
			return true
		})
		if err != nil {
			return nil, err
		}
		for n := len(hourlyLogs) - 1; n >= 0; n-- {
			logs = append(logs, hourlyLogs[n])
			if len(logs) >= limit {
				break
			}
		}
		if len(logs) >= limit {
			break
		}
	}
	return logs, nil
}

func (s *slowQueryV2) readSlowLogsOrderBy(
	ctx context.Context,
	store storage.Storage,
	bucket string,
	params *ListParams,
	begin, end time.Time,
	beginHour, endHour time.Time,
	limit int,
) ([]*SlowLog, error) {
	wg := &sync.WaitGroup{}
	logCh := make(chan *SlowLog)
	errCh := make(chan error)
	for cur := beginHour; cur.Before(endHour); cur = cur.Add(time.Hour) {
		hour := cur.UTC().Format(hourlyTimeFormat)
		wg.Add(1)
		go func(hour string) {
			defer wg.Done()
			err := s.readSlowLogsHour(ctx, store, bucket, params, begin, end, hour, func(sl *SlowLog) bool {
				logCh <- sl
				return true
			})
			if err != nil {
				errCh <- err
			}
		}(hour)
	}

	doneCh := make(chan struct{})
	go func() {
		wg.Wait()
		close(doneCh)
	}()

	h := newSlowLogsHeap(strings.ToLower(params.OrderBy), params.IsDesc, limit)
	for {
		select {
		case <-doneCh:
			return h.Slice(), nil
		case err := <-errCh:
			return nil, err
		case sl := <-logCh:
			h.Push(sl)
		}
	}
}

func (s *slowQueryV2) readSlowLogsHour(
	ctx context.Context,
	store storage.Storage,
	bucket string,
	params *ListParams,
	begin, end time.Time,
	hour string,
	f func(sl *SlowLog) bool,
) error {
	// key := fmt.Sprintf("0/%s/%s/%s/logs/slowlog-merged/hourly/data/%s.gz",
	// 	params.TenantID, params.ProjectID, params.ClusterID, hour)
	key := fmt.Sprintf("diagnosis/data/%s/slowlog-merged/1h/%s", params.ClusterID, hour)
	object, err := store.Object(ctx, bucket, key)
	if err != nil {
		var t *types.NoSuchKey
		if errors.As(err, &t) {
			return nil
		}
		if err == gcs.ErrObjectNotExist {
			return nil
		}
		var azerr *azcore.ResponseError
		if errors.As(err, &azerr) && azerr.StatusCode == http.StatusNotFound {
			return nil
		}
		return err
	}
	defer object.Reader.Close()
	reader := bufio.NewReader(object.Reader)
	for {
		line, err := readLine(reader)
		if err != nil {
			if err == io.EOF {
				break
			}
			return err
		}
		if line[0] == '#' { // comment line
			continue
		}
		if len(line) < 16 {
			continue
		}
		ts, err := strconv.ParseInt(string(line[:16]), 10, 64)
		if err != nil {
			return err
		}
		tm := time.UnixMicro(ts)
		if tm.Before(begin) {
			continue
		}
		if tm.After(end) || tm.Equal(end) {
			break
		}
		var sl SlowLog
		if err := json.Unmarshal(line[16:], &sl); err != nil {
			return err
		}
		sl.Time = tm
		if isInternalSlowLog(&sl, params.SharedTier) {
			// Internal slow queries are hidden unless explicitly requested.
			if !params.ShowInternal {
				continue
			}
		}
		if params.Text != "" {
			if !strings.Contains(sl.Query, strings.ToLower(params.Text)) &&
				!strings.Contains(sl.Query, strings.ToUpper(params.Text)) {
				continue
			}
		}
		if params.Digest != "" {
			if !strings.EqualFold(params.Digest, sl.Digest) {
				continue
			}
		}
		if len(params.DB) > 0 {
			matchDB := false
			for _, db := range params.DB {
				if sl.DB == db {
					matchDB = true
					break
				}
			}
			if !matchDB {
				continue
			}
		}
		if len(params.Plans) > 0 {
			found := false
			for _, plan := range params.Plans {
				if plan == sl.PlanDigest {
					found = true
					break
				}
			}
			if !found {
				continue
			}
		}
		if !f(&sl) {
			break
		}
	}
	return nil
}

func (s *slowQueryV2) Detail(ctx context.Context, params *DetailParams) (*SlowLog, error) {
	bucket := fmt.Sprintf("o11y-prod-shared-%s", params.Region)
	store, err := s.stores.Storage(ctx, params.Provider, params.Region)
	if err != nil {
		return nil, err
	}

	sec := int64(params.Timestamp)
	tmHour := time.Unix(sec-(sec%3600), 0)
	hour := tmHour.UTC().Format(hourlyTimeFormat)

	// key := fmt.Sprintf("0/%s/%s/%s/logs/slowlog-merged/hourly/data/%s.gz",
	// 	params.TenantID, params.ProjectID, params.ClusterID, hour)
	key := fmt.Sprintf("diagnosis/data/%s/slowlog-merged/1h/%s", params.ClusterID, hour)
	object, err := store.Object(ctx, bucket, key)
	if err != nil {
		return nil, err
	}
	defer object.Reader.Close()
	reader := bufio.NewReader(object.Reader)
	for {
		line, err := readLine(reader)
		if err != nil {
			if err == io.EOF {
				break
			}
			return nil, err
		}
		if line[0] == '#' { // comment line
			continue
		}
		if len(line) < 16 {
			continue
		}
		ts, err := strconv.ParseInt(string(line[:16]), 10, 64)
		if err != nil {
			return nil, err
		}
		tm := time.UnixMicro(ts)
		ft := timeToFloat(tm)
		// if params.Timestamp < ft {
		// 	break
		// }
		if params.Timestamp == ft {
			var sl SlowLog
			if err := json.Unmarshal(line[16:], &sl); err != nil {
				return nil, err
			}
			if params.ConnID == sl.ConnID && params.Digest == sl.Digest {
				if isInternalSlowLog(&sl, params.SharedTier) && !params.ShowInternal {
					continue
				}
				sl.Time = tm
				return &sl, nil
			}
		}
	}
	return nil, errors.New("not found")
}

func (s *slowQueryV2) Analyze(_ context.Context, _ *AnalyzeParams) error {
	return nil
}

func (s *slowQueryV2) AvailableFields(_ context.Context) ([]string, error) {
	return nil, nil
}

func (s *slowQueryV2) Download(_ context.Context, _ *DownloadParams) (*DownloadData, error) {
	return nil, nil
}

func stringTsToSecNsec(ts string) (int64, int64, error) {
	arr := strings.Split(ts, ".")
	if len(arr) != 2 {
		return 0, 0, errors.New("invalid timestamp")
	}
	sec, err := strconv.ParseInt(arr[0], 10, 64)
	if err != nil {
		return 0, 0, err
	}
	arr1 := arr[1]
	if len(arr1) < 9 {
		for n := 0; n < 9-len(arr1); n++ {
			arr[1] += "0"
		}
	} else if len(arr[1]) > 9 {
		arr[1] = arr[1][:9]
	}
	nsec, err := strconv.ParseInt(arr[1], 10, 64)
	if err != nil {
		return 0, 0, err
	}
	return sec, nsec, nil
}

//lint:ignore U1000 Ignore unused function
func stringTsToTime(ts string) (time.Time, error) {
	sec, nsec, err := stringTsToSecNsec(ts)
	if err != nil {
		return time.Time{}, err
	}
	return time.Unix(sec, nsec), nil
}