package client

import (
	"context"
	"os"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/juju/errors"
	"github.com/ngaut/log"
	"github.com/pingcap/schrodinger/pkg/elasticsearch"
	"github.com/pingcap/schrodinger/pkg/types"
	"github.com/pingcap/schrodinger/util"
)

const (
	managerAddrEnv  = "MANAGER_ADDR"
	boxIDEnv        = "BOX_ID"
	experimentIDEnv = "EXPERIMENT_ID"
	testIDEnv       = "TEST_ID"
	tidbServiceEnv  = "TIDB_SERVICE"
	pdServiceEnv    = "PD_SERVICE"
	tidbsEnv        = "TIDBS"
	pdsEnv          = "PDS"
	tikvsEnv        = "TIKVS"
	// KVName is tikv name
	KVName = "tikv"
	// DBName is tidb name
	DBName = "tidb"
	// PDName is pd name
	PDName = "pd"
)

// Mode is which mode client in
type Mode string

// Client Modes
const (
	ModeDev    Mode = "dev"
	ModeOnline      = "online"
)

// Client is used to communicate with the tidb-cluster and daemon for cases
type Client struct {
	sync.RWMutex
	ctx    context.Context
	cancel context.CancelFunc

	mode      Mode
	inited    uint32
	createCat bool

	conf *Config
	cat  *types.Cat

	selfID int64

	nemesis    *Nemesis
	failpoints *Failpoints

	managerAddr  string
	boxID        int64
	experimentID int64

	logClient *elasticsearch.Client
}

// New creates new client
func New(ctx context.Context, mode Mode) *Client {
	ctx, cancel := context.WithCancel(ctx)
	cli := &Client{
		ctx:       ctx,
		cancel:    cancel,
		mode:      mode,
		createCat: false,
		inited:    0,
		conf:      &Config{},
		nemesis:   &Nemesis{},
	}

	if err := cli.initConfig(); err != nil {
		Fatalf("fail to init config, error : %v", err)
	}

	return cli
}

// SelfInfo returns self test info
func (c *Client) SelfInfo() *TestInfo {
	if err := c.refreshConfig(); err != nil {
		Fatalf("init client config error %v", errors.ErrorStack(err))
	}
	return c.conf.Self
}

// Config returns self config
func (c *Client) Config() *Config {
	if err := c.refreshConfig(); err != nil {
		Fatalf("init client config error %v", errors.ErrorStack(err))
	}
	return c.conf
}

func (c *Client) refreshConfig() error {
	return errors.Trace(c.initConfig())
}

func (c *Client) initConfig() error {
	c.Lock()
	defer c.Unlock()

	var err error
	c.managerAddr = os.Getenv(managerAddrEnv)

	if boxID := os.Getenv(boxIDEnv); !util.IsEmpty(boxID) {
		c.boxID, err = strconv.ParseInt(boxID, 10, 64)
		if err != nil {
			return errors.Trace(err)
		}
	}

	if experimentID := os.Getenv(experimentIDEnv); !util.IsEmpty(experimentID) {
		c.experimentID, err = strconv.ParseInt(experimentID, 10, 64)
		if err != nil {
			return errors.Trace(err)
		}
	}

	if testIDStr := os.Getenv(testIDEnv); !util.IsEmpty(testIDStr) {
		c.selfID, err = strconv.ParseInt(testIDStr, 10, 64)
		if err != nil {
			return errors.Trace(err)
		}
	}

	switch c.mode {
	case ModeOnline:
		err = c.initOnlineConfig()
	case ModeDev:
		err = c.initDevConfig()
	default:
		return errors.NotSupportedf("mode :%s", c.mode)
	}

	if err != nil {
		return errors.Trace(err)
	}

	if c.inited == 0 {
		atomic.SwapUint32(&c.inited, uint32(1))
	}

	return nil
}

func (c *Client) initOnlineConfig() error {
	err := util.RetryOnError2(c.ctx, 20, 30*time.Second, func() (error, bool) {
		conf, err := getConfig(c.managerAddr, c.selfID)
		if err != nil && util.CheckIsNetworkError(err) {
			if c.inited == 1 && c.conf != nil {
				log.Warnf("get config failed: %v, just use the old config", err)
				return nil, true
			}

			return errors.Trace(err), false
		}

		if err == nil {
			c.conf = conf
			return nil, true
		}

		log.Errorf("get config failed: %v", err)
		return errors.Trace(err), true
	})
	if err != nil {
		log.Errorf("get config failed: %v", err)
		return errors.Trace(err)
	}

	// if without cat. no need to get nemesis and failpoint
	if c.conf.Cat == nil {
		return nil
	}
	// init nemesis
	c.nemesis.init(c.genNodes())

	// TODO: manager failpoints
	failpoints := make([]*types.Failpoint, 0)
	c.failpoints = newFailPoints(failpoints)

	// TODO watch config
	// go c.watchConfig()
	return nil
}

func (c *Client) genNodes() ([]*Node, []*Node, []*Node) {
	pdNodes := genNodes(c.ctx, c.conf.Cat.PDs, PDName)
	tidbNodes := genNodes(c.ctx, c.conf.Cat.TiDBs, DBName)
	tikvNodes := genNodes(c.ctx, c.conf.Cat.TiKVs, KVName)

	return pdNodes, tidbNodes, tikvNodes
}

// GetNodesWithStatus gets nodes with status
func (c *Client) GetNodesWithStatus(name string) ([]*Node, error) {
	if !c.isOnline() {
		Warnf("only mode online is permitted. the mode is %s", c.mode)
		return nil, nil
	}

	if err := c.refreshConfig(); err != nil {
		Fatalf("init client config error %v", errors.ErrorStack(err))
	}

	var nodes []*Node
	conf := c.conf.Cat
	switch name {
	case KVName:
		nodes = genNodes(c.ctx, conf.TiKVs, name)
	case DBName:
		nodes = genNodes(c.ctx, conf.TiDBs, name)
	case PDName:
		nodes = genNodes(c.ctx, conf.PDs, name)
	default:
		return nodes, errors.New("Node name not support")
	}
	return nodes, nil
}

func (c *Client) initDevConfig() error {
	tidbServiceStr := strings.TrimSpace(os.Getenv(tidbServiceEnv))
	pdServiceStr := strings.TrimSpace(os.Getenv(pdServiceEnv))
	tidbStr := strings.TrimSpace(os.Getenv(tidbsEnv))
	pdStr := strings.TrimSpace(os.Getenv(pdsEnv))
	tikvStr := strings.TrimSpace(os.Getenv(tikvsEnv))

	catConf, err := genFakeCatInfo(tidbServiceStr, pdServiceStr, tidbStr, pdStr, tikvStr)
	if err != nil {
		return errors.Trace(err)
	}

	c.conf = &Config{
		Cat: catConf,
	}

	return nil
}

// SetCreateCat sets create cat with cat config
func (c *Client) SetCreateCat(cat *types.Cat) {
	if !c.isOnline() {
		Warnf("only mode online is permitted. the mode is %s", c.mode)
		return
	}

	if err := c.refreshConfig(); err != nil {
		Fatalf("init client config error %v", errors.ErrorStack(err))
	}

	c.cat = cat
	c.createCat = true
}

func (c *Client) isDev() bool {
	return c.mode == ModeDev
}

func (c *Client) isOnline() bool {
	return c.mode == ModeOnline
}