package ether

import (
	"path"
	"strings"
	"sync/atomic"
	"time"

	"github.com/coreos/etcd/clientv3"
	"github.com/coreos/etcd/mvcc/mvccpb"
	"github.com/juju/errors"
	"github.com/ngaut/log"
	"github.com/pingcap/wormhole/pkg/whpb"
	"golang.org/x/net/context"
)

var (
	errNoLeader = errors.New("no leader")
)

// IsLeader returns whether server is leader or not.
func (s *Server) IsLeader() bool {
	return atomic.LoadInt64(&s.isLeaderValue) == 1
}

func (s *Server) enableLeader(b bool) {
	value := int64(0)
	if b {
		value = 1
	}

	atomic.StoreInt64(&s.isLeaderValue, value)
}

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

func (s *Server) leaderLoop() {
	s.wg.Add(1)
	defer s.wg.Done()

	for {
		if s.isClosed() {
			log.Infof("[ether] server is closed, return leader loop")
			return
		}

		leader, err := getLeader(s.client, s.getLeaderPath())
		if err != nil {
			log.Errorf("[ether] get leader err %v", err)
			time.Sleep(200 * time.Millisecond)
			continue
		}
		if leader != nil {
			if s.isSameLeader(leader) {
				// oh, we are already leader, we may meet something wrong
				// in previous campaignLeader. we can resign and campaign again.
				log.Warnf("[ether] leader is still %s, resign and campaign again", leader)
				if err = s.resignLeader(); err != nil {
					log.Errorf("[ether] resign leader err %s", err)
					time.Sleep(200 * time.Millisecond)
					continue
				}
			} else {
				log.Infof("[ether] leader is %s, watch it", leader)
				s.watchLeader()
				log.Info("[ether] leader changed, try to campaign leader")
			}
		}

		if err = s.campaignLeader(); err != nil {
			log.Errorf("[ether] campaign leader err %s", errors.ErrorStack(err))
		}
	}
}

func getLeaderAddr(leader *whpb.Member) string {
	return strings.Join(leader.GetClientUrls(), ",")
}

// getLeader gets server leader from etcd.
func getLeader(c *clientv3.Client, leaderPath string) (*whpb.Member, error) {
	leader := &whpb.Member{}
	ok, err := getProtoMsg(c, leaderPath, leader)
	if err != nil {
		return nil, errors.Trace(err)
	}
	if !ok {
		return nil, nil
	}

	return leader, nil
}

// GetLeader gets pd cluster leader.
func (s *Server) GetLeader() (*whpb.Member, error) {
	if s.isClosed() {
		return nil, errors.New("[ether] server is closed")
	}
	leader, err := getLeader(s.client, s.getLeaderPath())
	if err != nil {
		return nil, errors.Trace(err)
	}
	if leader == nil {
		return nil, errors.Trace(errNoLeader)
	}
	return leader, nil
}

func (s *Server) isSameLeader(leader *whpb.Member) bool {
	return leader.GetMemberId() == s.ID()
}

func (s *Server) marshalLeader() string {
	leader := &whpb.Member{
		Name:       s.Name(),
		MemberId:   s.ID(),
		ClientUrls: strings.Split(s.cfg.AdvertiseClientUrls, ","),
		PeerUrls:   strings.Split(s.cfg.AdvertisePeerUrls, ","),
	}

	data, err := leader.Marshal()
	if err != nil {
		// can't fail, so panic here.
		log.Fatalf("[ether] marshal leader %s err %v", leader, err)
	}

	return string(data)
}

func (s *Server) campaignLeader() error {
	log.Debugf("[ether] begin to campaign leader %s", s.Name())

	lessor := clientv3.NewLease(s.client)
	defer lessor.Close()

	start := time.Now()
	ctx, cancel := context.WithTimeout(s.client.Ctx(), requestTimeout)
	leaseResp, err := lessor.Grant(ctx, s.cfg.LeaderLease)
	cancel()

	if cost := time.Now().Sub(start); cost > slowRequestTime {
		log.Warnf("[ether] lessor grants too slow, cost %s", cost)
	}

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

	leaderKey := s.getLeaderPath()
	// The leader key must not exist, so the CreateRevision is 0.
	resp, err := s.txn().
		If(clientv3.Compare(clientv3.CreateRevision(leaderKey), "=", 0)).
		Then(clientv3.OpPut(leaderKey, s.leaderValue, clientv3.WithLease(clientv3.LeaseID(leaseResp.ID)))).
		Commit()
	if err != nil {
		return errors.Trace(err)
	}
	if !resp.Succeeded {
		return errors.New("[ether] campaign leader failed, other server may campaign ok")
	}

	// Make the leader keepalived.
	ctx, cancel = context.WithCancel(s.client.Ctx())
	defer cancel()

	ch, err := lessor.KeepAlive(ctx, clientv3.LeaseID(leaseResp.ID))
	if err != nil {
		return errors.Trace(err)
	}
	log.Debugf("[ether] campaign leader ok %s", s.Name())

	// Try to create cluster.
	err = s.createCluster()
	if err != nil {
		return errors.Trace(err)
	}
	defer s.stopCluster()

	s.enableLeader(true)
	defer s.enableLeader(false)

	log.Infof("[ether] ether cluster leader %s is ready to serve", s.Name())

	for {
		select {
		case _, ok := <-ch:
			if !ok {
				log.Info("[ether] keep alive channel is closed")
				return nil
			}
		case <-ctx.Done():
			return errors.New("[ether] server closed")
		}
	}
}

func (s *Server) watchLeader() {
	watcher := clientv3.NewWatcher(s.client)
	defer watcher.Close()

	ctx, cancel := context.WithCancel(s.client.Ctx())
	defer cancel()

	for {
		rch := watcher.Watch(ctx, s.getLeaderPath())
		for wresp := range rch {
			if wresp.Canceled {
				return
			}

			for _, ev := range wresp.Events {
				if ev.Type == mvccpb.DELETE {
					log.Info("[ether] leader is deleted")
					return
				}
			}
		}

		select {
		case <-ctx.Done():
			// server closed, return
			return
		default:
		}
	}
}

func (s *Server) resignLeader() error {
	// delete leader itself and let others start a new election again.
	leaderKey := s.getLeaderPath()
	resp, err := s.leaderTxn().Then(clientv3.OpDelete(leaderKey)).Commit()
	if err != nil {
		return errors.Trace(err)
	}
	if !resp.Succeeded {
		return errors.New("[ether] resign leader failed, we are not leader already")
	}

	return nil
}

func (s *Server) leaderCmp() clientv3.Cmp {
	return clientv3.Compare(clientv3.Value(s.getLeaderPath()), "=", s.leaderValue)
}