package ether

import (
	"math/rand"
	"net/http"
	"path"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"google.golang.org/grpc"

	"github.com/coreos/etcd/clientv3"
	"github.com/coreos/etcd/embed"
	"github.com/coreos/etcd/pkg/types"
	"github.com/juju/errors"
	"github.com/ngaut/log"
	"github.com/pingcap/wormhole/pkg/etcdutil"
	"github.com/pingcap/wormhole/pkg/whpb"
)

const (
	etcdTimeout     = time.Second * 3
	requestTimeout  = etcdutil.DefaultRequestTimeout
	slowRequestTime = etcdutil.DefaultSlowRequestTime

	etherRootPath      = "/ether"
	etherAPIPrefix     = "/ether/"
	etherRPCPrefix     = "/ether/rpc"
	etherClusterIDPath = "/ether/cluster_id"
)

func init() {
	rand.Seed(time.Now().UnixNano())
}

// Server is the ether server
type Server struct {
	cfg *Config

	etcd *embed.Etcd

	client *clientv3.Client

	clusterID uint64

	rootPath string

	kv *kv

	// for id allocator, we can use one allocator for
	// store, region and peer, because we just need
	// a unique ID.
	idAlloc *idAllocator

	isLeaderValue int64
	// leader value saved in etcd leader key.
	// Every write will use this to check leader validation.
	leaderValue string

	cluster *Cluster
	handler *Handler

	wg     sync.WaitGroup
	closed int64

	id uint64

	hbStreams *heartbeatStreams
}

// NewServer creates the ether server with given configuration.
func NewServer(cfg *Config) (*Server, error) {
	s := CreateServer(cfg)
	if err := s.StartEtcd(nil); err != nil {
		s.Close()
		return nil, errors.Trace(err)
	}

	return s, nil
}

// CreateServer creates the UNINITIALIZED ether server with given configuration.
func CreateServer(cfg *Config) *Server {
	log.Infof("Ether config - %v", cfg)
	rand.Seed(time.Now().UnixNano())

	s := &Server{
		cfg:           cfg,
		isLeaderValue: 0,
		closed:        1,
	}
	s.handler = newHandler(s)
	return s
}

// StartEtcd starts an embed etcd server with an user handler.
func (s *Server) StartEtcd(apiHandler http.Handler) error {
	etcdCfg, err := s.cfg.genEmbedEtcdConfig()
	if err != nil {
		return errors.Trace(err)
	}
	if apiHandler != nil {
		etcdCfg.UserHandlers = map[string]http.Handler{
			etherAPIPrefix: apiHandler,
		}
	}
	etcdCfg.ServiceRegister = func(gs *grpc.Server) { whpb.RegisterEtherServer(gs, s) }

	log.Info("[ether] start embed etcd")

	etcd, err := embed.StartEtcd(etcdCfg)
	if err != nil {
		return errors.Trace(err)
	}

	// Check cluster ID
	urlmap, err := types.NewURLsMap(s.cfg.InitialCluster)
	if err != nil {
		return errors.Trace(err)
	}
	if err = etcdutil.CheckClusterID(etcd.Server.Cluster().ID(), urlmap); err != nil {
		return errors.Trace(err)
	}

	endpoints := []string{etcdCfg.ACUrls[0].String()}

	log.Infof("[ether] create etcd v3 client with endpoints %v", endpoints)
	client, err := clientv3.New(clientv3.Config{
		Endpoints:   endpoints,
		DialTimeout: etcdTimeout,
	})
	if err != nil {
		return errors.Trace(err)
	}

	if err = etcdutil.WaitEtcdStart(client, endpoints[0]); err != nil {
		// See https://github.com/coreos/etcd/issues/6067
		// Here may return "not capable" error because we don't start
		// all etcds in initial_cluster at same time, so here just log
		// an error.
		// Note that ether can not work correctly if we don't start all etcds.
		log.Errorf("[ether] etcd start failed, err %v", err)
	}

	s.etcd = etcd
	s.client = client
	s.id = uint64(etcd.Server.ID())

	// update advertise peer urls.
	etcdMembers, err := etcdutil.ListEtcdMembers(client)
	if err != nil {
		return errors.Trace(err)
	}
	for _, m := range etcdMembers.Members {
		if s.ID() == m.ID {
			etcdPeerURLs := strings.Join(m.PeerURLs, ",")
			if s.cfg.AdvertisePeerUrls != etcdPeerURLs {
				log.Infof("[ether] update advertise peer urls from %s to %s", s.cfg.AdvertisePeerUrls, etcdPeerURLs)
				s.cfg.AdvertisePeerUrls = etcdPeerURLs
			}
		}
	}

	if err = s.initClusterID(); err != nil {
		return errors.Trace(err)
	}
	log.Infof("[ether] init cluster id %v", s.clusterID)

	s.rootPath = path.Join(etherRootPath, strconv.FormatUint(s.clusterID, 10))
	s.kv = newKV(s)
	s.hbStreams = newHeartbeatStreams(s.clusterID)
	s.idAlloc = &idAllocator{s: s}
	s.cluster = newCluster(s)

	// Server has started.
	atomic.StoreInt64(&s.closed, 0)
	return nil
}

func (s *Server) initClusterID() error {
	// Get any cluster key to parse the cluster ID.
	resp, err := etcdutil.KVGet(s.client, etherRootPath, clientv3.WithFirstCreate()...)
	if err != nil {
		return errors.Trace(err)
	}

	// If no key exist, generate a random cluster ID.
	if len(resp.Kvs) == 0 {
		s.clusterID, err = initOrGetClusterID(s.client, etherClusterIDPath)
		return errors.Trace(err)
	}

	key := string(resp.Kvs[0].Key)

	// If the key is "etherClusterIDPath", parse the cluster ID from it.
	if key == etherClusterIDPath {
		s.clusterID, err = bytesToUint64(resp.Kvs[0].Value)
		return errors.Trace(err)
	}

	// Parse the cluster ID from any other keys for compatibility.
	elems := strings.Split(key, "/")
	if len(elems) < 3 {
		return errors.Errorf("[ether] invalid cluster key %v", key)
	}
	s.clusterID, err = strconv.ParseUint(elems[2], 10, 64)

	log.Infof("[ether] init and load cluster id: %d", s.clusterID)
	return errors.Trace(err)
}

// Close closes the server.
func (s *Server) Close() {
	if !atomic.CompareAndSwapInt64(&s.closed, 0, 1) {
		// server is already closed
		return
	}

	log.Info("[ether] closing server")

	s.enableLeader(false)

	if s.client != nil {
		s.client.Close()
	}

	if s.etcd != nil {
		s.etcd.Close()
	}

	s.wg.Wait()

	log.Info("[ether] close server")
}

// isClosed checks whether server is closed or not.
func (s *Server) isClosed() bool {
	return atomic.LoadInt64(&s.closed) == 1
}

// ID returns the unique etcd ID for this server in etcd cluster.
func (s *Server) ID() uint64 {
	return s.id
}

// GetClient returns etcd client
func (s *Server) GetClient() *clientv3.Client {
	return s.client
}

// GetHandler returns the handler for API.
func (s *Server) GetHandler() *Handler {
	return s.handler
}

// Name returns the unique etcd Name for this server in etcd cluster.
func (s *Server) Name() string {
	return s.cfg.Name
}

// txn returns an etcd client transaction wrapper.
// The wrapper will set a request timeout to the context and log slow transactions.
func (s *Server) txn() clientv3.Txn {
	return etcdutil.NewSlowLogTxn(s.client)
}

// leaderTxn returns txn() with a leader comparison to guarantee that
// the transaction can be executed only if the server is leader.
func (s *Server) leaderTxn(cs ...clientv3.Cmp) clientv3.Txn {
	return s.txn().If(append(cs, s.leaderCmp())...)
}

// Run runs the ether server.
func (s *Server) Run() {
	// We use "127.0.0.1:0" for test and will set correct listening
	// address before run, so we set leader value here.
	s.leaderValue = s.marshalLeader()

	s.leaderLoop()
}

func (s *Server) getClusterRootPath() string {
	return path.Join(s.rootPath, "cluster")
}

// GetCluster gets cluster.
// If cluster has not been bootstrapped, return nil.
func (s *Server) GetCluster() *Cluster {
	if s.isClosed() || !s.cluster.isRunning() {
		return nil
	}
	return s.cluster
}

func (s *Server) createCluster() error {
	if s.cluster.isRunning() {
		return nil
	}

	return s.cluster.start()
}

func (s *Server) stopCluster() {
	s.cluster.stop()
}