package ether

import (
	"encoding/binary"
	"math"
	"math/rand"
	"strconv"
	"strings"
	"time"

	"github.com/coreos/etcd/clientv3"
	"github.com/coreos/etcd/mvcc/mvccpb"
	"github.com/gogo/protobuf/proto"
	"github.com/juju/errors"
	"github.com/pingcap/wormhole/pkg/etcdutil"
	"github.com/pingcap/wormhole/pkg/util"
	"github.com/pingcap/wormhole/pkg/whpb"
	"golang.org/x/net/context"
)

// A helper function to get value with key from etcd.
// TODO: return the value revision for outer use.
func getValue(c *clientv3.Client, key string, opts ...clientv3.OpOption) ([]byte, error) {
	resp, err := etcdutil.KVGet(c, key, opts...)
	if err != nil {
		return nil, errors.Trace(err)
	}

	if n := len(resp.Kvs); n == 0 {
		return nil, nil
	} else if n > 1 {
		return nil, errors.Errorf("[ether] invalid get value resp %v, must only one", resp.Kvs)
	}

	return resp.Kvs[0].Value, nil
}

// Return boolean to indicate whether the key exists or not.
// TODO: return the value revision for outer use.
func getProtoMsg(c *clientv3.Client, key string, msg proto.Message, opts ...clientv3.OpOption) (bool, error) {
	value, err := getValue(c, key, opts...)
	if err != nil {
		return false, errors.Trace(err)
	}
	if value == nil {
		return false, nil
	}

	if err = proto.Unmarshal(value, msg); err != nil {
		return false, errors.Trace(err)
	}

	return true, nil
}

func initOrGetClusterID(c *clientv3.Client, key string) (uint64, error) {
	ctx, cancel := context.WithTimeout(c.Ctx(), requestTimeout)
	defer cancel()

	// Generate a random cluster ID.
	ts := uint64(time.Now().Unix())
	clusterID := (ts << 32) + uint64(rand.Uint32())
	value := uint64ToBytes(clusterID)

	// Multiple PDs may try to init the cluster ID at the same time.
	// Only one PD can commit this transaction, then other PDs can get
	// the committed cluster ID.
	resp, err := c.Txn(ctx).
		If(clientv3.Compare(clientv3.CreateRevision(key), "=", 0)).
		Then(clientv3.OpPut(key, string(value))).
		Else(clientv3.OpGet(key)).
		Commit()
	if err != nil {
		return 0, errors.Trace(err)
	}

	// Txn commits ok, return the generated cluster ID.
	if resp.Succeeded {
		return clusterID, nil
	}

	// Otherwise, parse the committed cluster ID.
	if len(resp.Responses) == 0 {
		return 0, errors.Errorf("[ether] txn returns empty response: %v", resp)
	}

	response := resp.Responses[0].GetResponseRange()
	if response == nil || len(response.Kvs) != 1 {
		return 0, errors.Errorf("[ether] txn returns invalid range response: %v", resp)
	}

	return bytesToUint64(response.Kvs[0].Value)
}

func watch(ctx context.Context, client *clientv3.Client, prefix string, f func(event *clientv3.Event) error) error {
	watcher := clientv3.NewWatcher(client)
	defer watcher.Close()
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()

	for {
		select {
		case <-ctx.Done():
			// server closed, return
			return errors.Trace(ctx.Err())
		default:
		}
		rch := watcher.Watch(ctx, prefix, clientv3.WithPrefix())
		for wresp := range rch {
			if wresp.Canceled {
				return nil
			}

			for _, ev := range wresp.Events {
				if err := f(ev); err != nil {
					return errors.Trace(err)
				}
			}
		}
	}
}

func bytesToUint64(b []byte) (uint64, error) {
	if len(b) != 8 {
		return 0, errors.Errorf("[ether] invalid data, must 8 bytes, but %d", len(b))
	}

	return binary.BigEndian.Uint64(b), nil
}

func uint64ToBytes(v uint64) []byte {
	b := make([]byte, 8)
	binary.BigEndian.PutUint64(b, v)
	return b
}

// GetMembers return a slice of Members.
func GetMembers(etcdClient *clientv3.Client) ([]*whpb.Member, error) {
	listResp, err := etcdutil.ListEtcdMembers(etcdClient)
	if err != nil {
		return nil, errors.Trace(err)
	}

	members := make([]*whpb.Member, 0, len(listResp.Members))
	for _, m := range listResp.Members {
		info := &whpb.Member{
			Name:       m.Name,
			MemberId:   m.ID,
			ClientUrls: m.ClientURLs,
			PeerUrls:   m.PeerURLs,
		}
		members = append(members, info)
	}

	return members, nil
}

// make partitionsBitmap be a bitmap, every bit corresponds to one partition
// partitionNum contains a partition set [1,2,...,partitionNum]
func partitionsBitmap(partitionNum int32) []uint64 {
	if partitionNum == 0 {
		return nil
	}

	bitMapLength := int((partitionNum + 63) / 64)
	bitmap := make([]uint64, bitMapLength)

	for i := 0; i < bitMapLength-1; i++ {
		bitmap[i] = math.MaxUint64
	}

	// handle remain bits
	remainBitNum := int(partitionNum) - (bitMapLength-1)*64
	bitmap[bitMapLength-1] = 1<<uint64(remainBitNum) - 1

	return bitmap
}

func minDuration(a, b time.Duration) time.Duration {
	if a < b {
		return a
	}
	return b
}

// extendLease let worker have enough time to renew lease
func extendLease(lease int64) int64 {
	return lease * 3 / 2
}

func parseContractKey(item *mvccpb.KeyValue) (jobID uint64, workerID uint64, err error) {
	keyArrayLength := 7 // ref: /ether/{cluster_id}/cluster/contracts/jobID/workerID

	key := string(item.Key)
	strs := strings.Split(key, "/")
	if len(strs) < keyArrayLength {
		err = errors.Errorf("[ether] error key %s", key)
	}

	workerID, err = strconv.ParseUint(strs[len(strs)-1], 10, 64)
	if err != nil {
		return
	}
	jobID, err = strconv.ParseUint(strs[len(strs)-2], 10, 64)
	return jobID, workerID, errors.Trace(err)
}

func parsePipelineKey(item *mvccpb.KeyValue) (jobID uint64, err error) {
	keyArrayLength := 6 // ref: /ether/{cluster_id}/cluster/pipelines/jobID

	key := string(item.Key)
	strs := strings.Split(key, "/")
	if len(strs) < keyArrayLength {
		err = errors.Errorf("[ether] error key %s", key)
	}

	jobID, err = strconv.ParseUint(strs[len(strs)-1], 10, 64)
	return jobID, errors.Trace(err)
}

// -1 => s < t;
// 0 => s == t;
// 1 => s > t
func comparePosition(s, t *whpb.Position, tp whpb.SourceType) (int, error) {
	switch tp {
	case whpb.SourceType_mysql_binlog, whpb.SourceType_mysql_dump:
		return util.CompareBinlogPosition(s, t), nil
	default:
		return 0, errors.Errorf("[ether] not support tp %s comparison, position %v vs %v", tp, s, t)
	}
}

func compareKafakOffset(s, t *whpb.Position) int {
	sOffset := s.GetKafkaOffset()
	tOffset := t.GetKafkaOffset()

	if sOffset > tOffset {
		return 1
	} else if sOffset < tOffset {
		return -1
	}

	return 0
}