package conprof

import (
	"bytes"
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"io/ioutil"
	"net"
	"net/http"
	"net/url"
	"sort"
	"time"

	. "github.com/onsi/gomega"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/sdk/pkg/proxy"
	"github.com/pingcap/test-infra/sdk/resource"
	"github.com/pingcap/tidb-dashboard/pkg/utils/topology"
	"github.com/tidwall/gjson"
	"go.etcd.io/etcd/clientv3"
	"go.uber.org/zap"
	netproxy "golang.org/x/net/proxy"
	"google.golang.org/grpc"

	_ "github.com/pingcap/test-infra/sdk/resource/impl/k8s"

	"github.com/pingcap/endless/pkg/workloadnode"
	"github.com/pingcap/endless/pkg/workloadnode/tiup"
)

const DefaultVersion = "nightly"

var tiupTopoTmp = `
global:
  user: "tidb-%s"
  deploy_dir: "/tiup-%s/deploy"
  data_dir: "/tiup-%s/data"

pd_servers:
  - host: pd-peer
    client_port: %d
    peer_port: %d

tidb_servers:
  - host: tidb-peer
    port: %d
    status_port: %d

tikv_servers:
  - host: tikv-peer
    port: %d
    status_port: %d

monitoring_servers:
  - host: tiup-peer
    port: %d
    %s

%s
`

type ClusterConfig struct {
	Version              string
	Name                 string
	PDClientPort         uint16
	PDPeerPort           uint16
	TiDBPort             uint16
	TiDBStatusPort       uint16
	TiKVPort             uint16
	TiKVStatusPort       uint16
	PrometheusPort       uint16
	NodeExporterPort     uint16
	BlackboxExporterPort uint16
	NgmPort              uint16
}

func (c *ClusterConfig) buildTopologyYaml() string {
	ngmCfg := ""
	if c.NgmPort > 0 {
		ngmCfg = fmt.Sprintf("ng_port: %d", c.NgmPort)
	} else {
		ngmCfg = "ng_port: -1"
	}
	monitoredCfg := ""
	if c.NodeExporterPort > 0 {
		monitoredCfg += "monitored:\n"
		monitoredCfg += fmt.Sprintf("  node_exporter_port: %d\n", c.NodeExporterPort)
	}
	if c.BlackboxExporterPort > 0 {
		if monitoredCfg == "" {
			monitoredCfg += "monitored:\n"
		}
		monitoredCfg += fmt.Sprintf("  blackbox_exporter_port: %d\n", c.BlackboxExporterPort)
	}

	return fmt.Sprintf(tiupTopoTmp,
		c.Name, c.Name, c.Name,
		c.PDClientPort, c.PDPeerPort,
		c.TiDBPort, c.TiDBStatusPort,
		c.TiKVPort, c.TiKVStatusPort,
		c.PrometheusPort, ngmCfg, monitoredCfg)
}

type ClusterBundle struct {
	cfg ClusterConfig

	ctx    context.Context
	cancel context.CancelFunc

	tiupNode resource.WorkloadNode
	Cluster  *tiup.TiUPCluster
	DB       *sql.DB

	pdPorts map[uint16]struct{}
	etcdCli *clientv3.Client

	sysbenchNode resource.WorkloadNode
	sysbenchCmd  map[uint16]*workloadnode.Cmd
}

func NewClusterBundle(tiupNode, sysbenchNode resource.WorkloadNode, clusterCfg ClusterConfig) (b *ClusterBundle, err error) {
	if clusterCfg.Version == "" {
		clusterCfg.Version = DefaultVersion
	}
	ctx, cancel := context.WithCancel(context.Background())
	b = &ClusterBundle{
		cfg:          clusterCfg,
		ctx:          ctx,
		cancel:       cancel,
		tiupNode:     tiupNode,
		sysbenchNode: sysbenchNode,
		sysbenchCmd:  map[uint16]*workloadnode.Cmd{},
		pdPorts:      map[uint16]struct{}{clusterCfg.PDClientPort: {}},
		Cluster:      tiup.NewTiUPCluster(tiupNode, "tiup-peer", clusterCfg.buildTopologyYaml(), clusterCfg.Name),
	}
	if err = b.Cluster.Deploy(b.cfg.Version); err != nil {
		return
	}
	if err = b.Cluster.Start(); err != nil {
		return
	}

	b.DB, err = b.getDB("", clusterCfg.TiDBPort)
	if err != nil {
		return
	}
	b.DB.SetConnMaxLifetime(time.Minute * 3)
	b.DB.SetMaxOpenConns(10)
	b.DB.SetMaxIdleConns(10)

	err = b.initEtcdClient()
	return
}

func (b *ClusterBundle) NgmURL(ngPort uint16) string {
	return fmt.Sprintf("http://tiup-peer.%s:%d", b.Cluster.TestBed, ngPort)
}

func (b *ClusterBundle) NgmTopology() (string, error) {
	if b.etcdCli == nil {
		return "", errors.New("etcd client is nil")
	}
	return topology.FetchNgMonitoringTopology(b.ctx, b.etcdCli)
}

func (b *ClusterBundle) PrepareSysbench(tidbPort uint16) error {
	if b.DB == nil {
		return errors.New("DB is nil")
	}

	_, err := b.DB.ExecContext(b.ctx, "CREATE DATABASE sbtest")
	if err != nil {
		return err
	}

	return workloadnode.CommandContext(b.ctx, fmt.Sprintf(`/usr/bin/sysbench \
		--db-driver=mysql \
		--mysql-host=%s \
		--mysql-port=%d \
		--mysql-user=root \
		--tables=10 \
		--table-size=10000 \
		oltp_point_select \
		prepare`, "tidb-peer", tidbPort)).Run(b.sysbenchNode)
}

func (b *ClusterBundle) DestroySysbench() error {
	if b.DB == nil {
		return errors.New("DB is nil")
	}

	_, err := b.DB.ExecContext(b.ctx, "DROP DATABASE sbtest")
	return err
}

func (b *ClusterBundle) StartSysbench(tidbPort uint16) error {
	if b.DB == nil {
		return errors.New("DB is nil")
	}

	if _, ok := b.sysbenchCmd[tidbPort]; ok {
		return errors.New("sysbench is running")
	}

	cmd := workloadnode.CommandContext(b.ctx, fmt.Sprintf(`/usr/bin/sysbench \
		--db-driver=mysql \
		--mysql-host=%s \
		--mysql-port=%d \
		--mysql-user=root \
		--tables=10 \
		--table-size=10000 \
		--time=30000 \
		--threads=1 \
		oltp_point_select \
		run`, "tidb-peer", tidbPort))
	b.sysbenchCmd[tidbPort] = cmd
	return cmd.Start(b.sysbenchNode)
}

func (b *ClusterBundle) StopSysbench(tidbPort uint16) {
	cmd, ok := b.sysbenchCmd[tidbPort]
	if !ok {
		return
	}

	_ = cmd.Stop()
	_ = cmd.Wait()
	delete(b.sysbenchCmd, tidbPort)
}

func (b *ClusterBundle) StopAllSysbench() {
	for k, v := range b.sysbenchCmd {
		_ = v.Stop()
		_ = v.Wait()
		delete(b.sysbenchCmd, k)
	}
}

func (b *ClusterBundle) ScaleOut(component string, portFirst, portSecond uint16) error {
	scaleOutTmp := ""
	switch component {
	case "tidb":
		scaleOutTmp = `
tidb_servers:
  - host: tidb-peer
    port: %d
    status_port: %d
`
	case "tikv":
		scaleOutTmp = `
tikv_servers:
  - host: tikv-peer
    port: %d
    status_port: %d
`
	case "pd":
		scaleOutTmp = `
pd_servers:
  - host: pd-peer
    client_port: %d
    peer_port: %d
`
	case "monitoring":
		scaleOutTmp = `
monitoring_servers:
  - host: tiup-peer
    port: %d
    ng_port: %d
`
	default:
		return errors.New("component is valid for tidb, tikv, pd or monitoring")
	}

	yml := fmt.Sprintf(scaleOutTmp, portFirst, portSecond)
	if err := b.Cluster.ScaleOut(yml); err != nil {
		return err
	}

	if component == "pd" {
		b.pdPorts[portFirst] = struct{}{}
		return b.initEtcdClient()
	}
	return nil
}

func (b *ClusterBundle) ScaleIn(component string, port uint16) error {
	// some pre-processing
	switch component {
	case "pd":
		// evict leader
		if err := workloadnode.CommandContext(b.ctx, fmt.Sprintf(`
tiup ctl:%s pd -u http://pd-peer:%d member leader show | grep %d -q && \
tiup ctl:%s pd -u http://pd-peer:%d member leader resign || \
true
`, b.cfg.Version, port, port, b.cfg.Version, port)).Run(b.tiupNode); err != nil {
			return err
		}
	case "tikv":
		// set replicas to 1 to avoid failure
		for pdPort := range b.pdPorts {
			if err := workloadnode.CommandContext(b.ctx, fmt.Sprintf(`
tiup ctl:%s pd -u http://pd-peer:%d config set max-replicas 1 || true
`, b.cfg.Version, pdPort)).Run(b.tiupNode); err != nil {
				return err
			}
		}
	}

	node := ""
	if component == "monitoring" {
		node = fmt.Sprintf("tiup-peer:%d", port)
	} else {
		node = fmt.Sprintf("%s-peer:%d", component, port)
	}
	if err := b.Cluster.ScaleIn(node); err != nil {
		return err
	}

	if component == "pd" {
		delete(b.pdPorts, port)
		return b.initEtcdClient()
	}

	return nil
}

func (b *ClusterBundle) StopComponent(component string, port uint16) error {
	node := component
	if component == "monitoring" {
		node = "tiup"
	}
	return b.Cluster.Stop(fmt.Sprintf("-N %s-peer:%d", node, port))
}

func (b *ClusterBundle) RestartComponent(component string, port uint16) error {
	return b.Cluster.Restart(fmt.Sprintf("-N %s-peer:%d", component, port))
}

func (b *ClusterBundle) Close() {
	_ = b.closeEtcd()
	b.StopAllSysbench()
	_ = b.DestroySysbench()
	_ = workloadnode.CommandContext(b.ctx, "pkill haproxy").Run(b.tiupNode)
	_ = workloadnode.CommandContext(b.ctx, fmt.Sprintf("tiup cluster destroy %s -y", b.cfg.Name)).Run(b.tiupNode)
	b.cancel()
}

func (b *ClusterBundle) getDB(dbName string, port uint16) (*sql.DB, error) {
	addr, err := url.Parse(fmt.Sprintf("http://tidb-peer.%s:%d", b.Cluster.TestBed, port))
	if err != nil {
		return nil, err
	}

	db, err := sql.Open("mysql", resource.BuildDSN(addr, dbName))
	if err != nil {
		return nil, err
	}

	if err = db.PingContext(b.ctx); err != nil {
		return nil, err
	}
	return db, nil
}

func (b *ClusterBundle) initEtcdClient() (err error) {
	if b.etcdCli != nil {
		if err = b.etcdCli.Close(); err != nil {
			return
		}
		b.etcdCli = nil
	}

	var endpoints []string
	for k := range b.pdPorts {
		endpoints = append(endpoints, fmt.Sprintf("pd-peer.%s:%d", b.Cluster.TestBed, k))
	}
	dialer := proxy.GrpcProxyDialer()
	zapCfg := zap.NewProductionConfig()
	zapCfg.Encoding = log.ZapEncodingName
	b.etcdCli, err = clientv3.New(clientv3.Config{
		Context:   b.ctx,
		Endpoints: endpoints,
		DialOptions: []grpc.DialOption{
			grpc.WithContextDialer(func(ctx context.Context, addr string) (net.Conn, error) {
				switch d := dialer.(type) {
				case netproxy.ContextDialer:
					return d.DialContext(ctx, "tcp", addr)
				default:
					return d.Dial("tcp", addr)
				}
			}),
		},
		LogConfig: &zapCfg,
	})

	return
}

func (b *ClusterBundle) closeEtcd() error {
	if b.etcdCli == nil {
		return nil
	}
	err := b.etcdCli.Close()
	b.etcdCli = nil
	return err
}

func (b *ClusterBundle) ngmConfigured() bool {
	return b.cfg.NgmPort > 0
}

func CheckTiUPAvailable(node resource.WorkloadNode) error {
	_, stderr, exitCode, err := node.Exec(resource.WorkloadNodeExecOptions{Command: "tiup update --self && tiup list"})
	if exitCode != 0 {
		return errors.New(stderr)
	}
	if err != nil {
		return err
	}
	return nil
}

func (b *ClusterBundle) TurnOnConprof(ngmURL string) error {
	cfg := []byte(`{"continuous_profiling": {"enable": true}}`)
	_, err := http.Post(fmt.Sprintf("%s/config", ngmURL), "application/json", bytes.NewBuffer(cfg))
	return err
}

func (b *ClusterBundle) TurnOffTopSQL() error {
	if b.DB == nil {
		return errors.New("DB is nil")
	}

	_, err := b.DB.ExecContext(b.ctx, "SET GLOBAL tidb_enable_top_sql = off")
	return err
}

func (b *ClusterBundle) ExpectConprofIsRunning(g Gomega, ngmURL string) {
	cfg, err := b.GetConprofCfg(ngmURL)
	g.Expect(err).ShouldNot(HaveOccurred())
	g.Expect(QueryJSON(cfg, "continuous_profiling.enable")).Should(Equal(true))
}

func (b *ClusterBundle) ExpectComponentsProfileData(g Gomega, ngmURL string, comps Components) {
	ts, err := b.GetConprofFinishedTs(ngmURL)
	g.Expect(err).ShouldNot(HaveOccurred())
	gpd, err := b.GetConprofProfileDataByTs(ngmURL, ts)
	g.Expect(err).ShouldNot(HaveOccurred())

	log.Info("check profile data", zap.Any("expect", comps), zap.Any("result", gpd.TargetProfiles))
	for _, comp := range comps {
		addr := fmt.Sprintf("%v:%v", comp.IP, comp.Port)
		found := false
		for _, pt := range gpd.TargetProfiles {
			if pt.Target.Component == comp.Name && pt.Target.Address == addr {
				found = true
				break
			}
		}
		g.Expect(found).Should(Equal(true), fmt.Sprintf("component profile data not find, componnent: %v, get: %v", comp, gpd.TargetProfiles))
	}
}

func (b *ClusterBundle) ExpectComponentsProfileDataCs(ngmURL string, cmps Components) {

}

func (b *ClusterBundle) GetConprofCfg(ngmURL string) ([]byte, error) {
	resp, err := http.Get(fmt.Sprintf("%s/config", ngmURL))
	if err != nil {
		return nil, err
	}
	return ioutil.ReadAll(resp.Body)
}

func (b *ClusterBundle) GetConprofComponent(ngmURL string) (Components, error) {
	resp, err := http.Get(fmt.Sprintf("%s/continuous_profiling/components", ngmURL))
	if err != nil {
		return nil, err
	}
	data, err := ioutil.ReadAll(resp.Body)
	components := Components{}
	if err != nil {
		return nil, err
	}
	err = json.Unmarshal(data, &components)
	if err != nil {
		return nil, err
	}
	sort.Sort(components)
	return components, err
}

func (b *ClusterBundle) GetConprofFinishedTs(ngmURL string) (int64, error) {
	t := time.Now()
	begin := t.Add(-time.Minute * 5).Unix()
	end := t.Unix()
	path := fmt.Sprintf("%s/continuous_profiling/group_profiles?begin_time=%v&end_time=%v", ngmURL, begin, end)
	resp, err := http.Get(path)
	if err != nil {
		return 0, err
	}
	data, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return 0, err
	}
	groupProfilesList := []GroupProfiles{}
	err = json.Unmarshal(data, &groupProfilesList)
	if err != nil {
		return 0, err
	}
	for _, gp := range groupProfilesList {
		if gp.State == "finished" {
			return gp.Ts, nil
		}
	}
	return 0, errors.New("doesn't have profile data")
}

func (b *ClusterBundle) GetConprofProfileDataByTs(ngmURL string, ts int64) (*GroupProfileDetail, error) {
	path := fmt.Sprintf("%s/continuous_profiling/group_profile/detail?ts=%v", ngmURL, ts)
	resp, err := http.Get(path)
	if err != nil {
		return nil, err
	}
	data, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return nil, err
	}
	groupProfileDetail := GroupProfileDetail{}
	err = json.Unmarshal(data, &groupProfileDetail)
	return &groupProfileDetail, err
}

type Component struct {
	Name       string `json:"name"`
	IP         string `json:"ip"`
	Port       uint   `json:"port"`
	StatusPort uint   `json:"status_port"`
}

type Components []Component

func (c Components) Len() int           { return len(c) }
func (c Components) Less(i, j int) bool { return fmt.Sprintf("%v", c[i]) < fmt.Sprintf("%v", c[j]) }
func (c Components) Swap(i, j int)      { c[i], c[j] = c[j], c[i] }

type GroupProfiles struct {
	Ts          int64  `json:"ts"`
	ProfileSecs int    `json:"profile_duration_secs"`
	State       string `json:"state"`
}

type GroupProfileDetail struct {
	Ts             int64           `json:"ts"`
	ProfileSecs    int             `json:"profile_duration_secs"`
	State          string          `json:"state"`
	TargetProfiles []ProfileDetail `json:"target_profiles"`
}

type ProfileDetail struct {
	State  string `json:"state"`
	Error  string `json:"error"`
	Type   string `json:"profile_type"`
	Target Target `json:"target"`
}

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

func QueryJSON(bytes []byte, path string) interface{} {
	return gjson.GetBytes(bytes, path).Value()
}