package large

import (
	"bytes"
	"fmt"
	"io"
	"net/http"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/VividCortex/gohistogram"
	"github.com/pingcap/errors"
)

type MemUsageRecord struct {
	HeapInUse    float64
	HeapSys      float64
	HeapReleased float64
	time.Time
}

type Report struct {
	Hist *gohistogram.NumericHistogram
	Succ int
	Fail int
	time.Duration
	Error    bool
	MemUsage []MemUsageRecord
}

func (r *Report) Report() {
	fmt.Println("========= hisgotram ==========")
	fmt.Println(r.Hist)
	fmt.Println("========= summary ==========")
	fmt.Println("Mean:", r.Hist.Mean())
	fmt.Println("Quantile 80:", r.Hist.Quantile(0.8))
	fmt.Println("Quantile 95:", r.Hist.Quantile(0.95))
	fmt.Println("Quantile 99:", r.Hist.Quantile(0.99))
	fmt.Println("Total Succ:", r.Succ)
	fmt.Println("Total Fail:", r.Fail)
	fmt.Println("LargeTxn Duration:", r.Duration)
	if r.Error {
		fmt.Println("Failed!!!")
	}
	if len(r.MemUsage) > 0 {
		real, rss := maxMem(r.MemUsage)
		fmt.Println("Mem Usage: ", real)
		fmt.Println("Mem RSS: ", rss)
	}
}

func maxMem(records []MemUsageRecord) (float64, float64) {
	real := records[0].HeapInUse
	rss := records[0].HeapSys - records[0].HeapReleased

	for i := 1; i < len(records); i++ {
		r := records[i]
		if r.HeapInUse > real {
			real = r.HeapInUse
		}

		if r.HeapSys-r.HeapReleased > rss {
			rss = r.HeapSys - r.HeapReleased
		}
	}

	return real, rss
}

type statsWorker struct {
	// small txn worker sends to this channel and statsWorker drain from it.
	// nil message means a small txn worker is exited.
	sampleCh    chan []float64
	workerCount int
	sync.WaitGroup
}

func newStatsWorker(workerCount int) *statsWorker {
	s := statsWorker{
		sampleCh:    make(chan []float64, 10),
		workerCount: workerCount,
	}
	s.WaitGroup.Add(1)
	return &s
}

func (stats *statsWorker) Run(hist *gohistogram.NumericHistogram) {
	finished := 0
	for data := range stats.sampleCh {
		if data == nil {
			// use data == nil to mean finish
			finished++
			if finished == stats.workerCount {
				break
			}
		}
		for _, val := range data {
			hist.Add(val)
		}
	}

	stats.WaitGroup.Done()
}

func memUseMonitor(done <-chan struct{}, metricURL string, report *Report) {
	tick := time.NewTicker(3 * time.Second)
	defer tick.Stop()

	for {
		select {
		case <-done:
			return
		case <-tick.C:
			r, err := getMemoryUsage(metricURL)
			if err != nil {
				fmt.Println(err)
				continue
			}
			report.MemUsage = append(report.MemUsage, r)
		}
	}
}

func getMemoryUsage(url string) (MemUsageRecord, error) {
	var record MemUsageRecord
	record.Time = time.Now()

	resp, err := http.Get(url)
	if err != nil {
		return record, errors.WithStack(err)
	}
	defer resp.Body.Close()

	var buf bytes.Buffer
	io.Copy(&buf, resp.Body)

	lines := strings.Split(buf.String(), "\n")
	for _, line := range lines {
		if strings.HasPrefix(line, "go_memstats_heap_inuse_bytes") {
			record.HeapInUse = parseMetric(line)
		} else if strings.HasPrefix(line, "go_memstats_heap_sys_bytes") {
			record.HeapSys = parseMetric(line)
		} else if strings.HasPrefix(line, "go_memstats_heap_released_bytes") {
			record.HeapReleased = parseMetric(line)
		}
	}

	return record, nil
}

// parseMetric get the float value from 'go_memstats_stack_inuse_bytes 1.736704e+06'
func parseMetric(str string) float64 {
	tmp := strings.Split(str, " ")
	if len(tmp) == 2 {
		f, err := strconv.ParseFloat(tmp[1], 64)
		if err == nil {
			return f
		}
	}
	return 0
}