package metrics

import (
	"bytes"
	"context"
	"fmt"
	"net/http"
	"os/exec"
	"strconv"
	"time"

	"github.com/golang/snappy"
	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"github.com/prometheus/prometheus/prompb"
	"go.uber.org/multierr"
	"go.uber.org/zap"
)

const (
	defaultMaxInsertRequestBytes = 33554430
	retryCount                   = 3
	retryInterval                = time.Second * 5
)

type MetricSink struct {
	Host                  string
	Port                  uint16
	ClusterModel          bool
	MaxInsertRequestBytes int
	OrgID                 uint
}

func (m *MetricSink) Sink(tss prompb.TimeSeries) error {
	wr := prompb.WriteRequest{
		Timeseries: []prompb.TimeSeries{tss},
	}

	data, err := wr.Marshal()
	if err != nil {
		return err
	}

	return m.sink(data)
}

// tsdbDirPath should be a abs path
func (m *MetricSink) SinkTSDBVmCtl(tsdbDirPath string, extLabels []string) error {
	vmInsertAddr := fmt.Sprintf("http://%s:%d", m.Host, m.Port)
	args := []string{
		"prometheus", "-s",
		"--prom-snapshot", tsdbDirPath,
		"--vm-addr", vmInsertAddr,
		"--prom-concurrency", "8",
		"--vm-account-id", strconv.FormatUint(uint64(m.OrgID), 10),
	}
	// label has format key=value
	for _, label := range extLabels {
		args = append(args, "--vm-extra-label", label)
	}
	cmd := exec.CommandContext(context.Background(), "/vmctl-prod", args...)
	bs, err := cmd.CombinedOutput()
	if err != nil {
		log.Error("exec vmctl failed", zap.String("cmd", cmd.String()), zap.String("output", string(bs)), zap.Error(err))
	}
	return err
}

func (m *MetricSink) SinkBatch(tss []prompb.TimeSeries) error {
	blocks, err := m.splitInsertData(tss)
	if err != nil {
		return err
	}
	var errs error
	for _, data := range blocks {
		errs = multierr.Append(errs, retry(retryCount, retryInterval, func() error {
			return m.sink(data)
		}))
	}

	return errs
}

func (m *MetricSink) getMaxInsertRequestBytes() int {
	if m.MaxInsertRequestBytes == 0 {
		return defaultMaxInsertRequestBytes
	}

	return m.MaxInsertRequestBytes
}

func (m *MetricSink) splitInsertData(tss []prompb.TimeSeries) ([][]byte, error) {
	if len(tss) == 0 {
		return nil, nil
	}

	wr := prompb.WriteRequest{
		Timeseries: tss,
	}

	data, err := wr.Marshal()
	if err != nil {
		return nil, err
	}

	if len(data) < m.getMaxInsertRequestBytes() {
		return [][]byte{data}, nil
	}

	if len(tss) == 1 {
		labelMap := make(map[string]string)
		for _, label := range tss[0].Labels {
			labelMap[label.Name] = label.Value
		}
		log.Warn("the single timeseries too big", zap.Int("actualSize", len(data)), zap.String("Name", labelMap["__name__"]), zap.String("Cluster", labelMap[ClusterLabel]), zap.String("host", labelMap[HostLabel]))
		return [][]byte{data}, nil
	}

	lresult, err := m.splitInsertData(tss[:len(tss)/2])
	if err != nil {
		return nil, err
	}

	rresult, err := m.splitInsertData(tss[len(tss)/2:])
	if err != nil {
		return nil, err
	}

	return append(lresult, rresult...), nil
}

var cli = &http.Client{
	Timeout: time.Duration(15) * time.Second,
	Transport: &http.Transport{
		MaxIdleConnsPerHost:   5,
		MaxConnsPerHost:       10,
		IdleConnTimeout:       90 * time.Second,
		TLSHandshakeTimeout:   10 * time.Second,
		ExpectContinueTimeout: 5 * time.Second,
	},
}

func (m *MetricSink) sink(data []byte) error {
	body := snappy.Encode(nil, data)
	uri := "/api/v1/write"
	if m.ClusterModel {
		uri = fmt.Sprintf("/insert/%d/prometheus%s", m.OrgID, uri)
	}

	req, err := http.NewRequest("POST", fmt.Sprintf("http://%s:%d%s", m.Host, m.Port, uri), bytes.NewReader(body))
	if err != nil {
		return err
	}
	req.Header.Set("Content-Type", "application/x-protobuf")

	resp, err := cli.Do(req)
	if err != nil {
		log.Error("request vm failed", zap.Error(err))
		return err
	}
	defer resp.Body.Close()

	if resp.StatusCode >= 400 {
		return errors.Errorf("send data to vm failed, err=%v", resp.Status)
	}

	return nil
}

func retry(attempts int, sleep time.Duration, f func() error) error {
	err := f()
	if err != nil {
		attempts--
		if attempts > 0 {
			log.Warn("retry func", zap.Int("attempts", attempts), zap.Error(err))
			time.Sleep(sleep)

			return retry(attempts, sleep, f)
		}

		return err
	}

	return nil
}