// GetVariable Gets a machine's variable, or the global if it was not
// set for the machine
func (m *etcdMachineInterface) GetVariable(key string) (string, error) {
	value, err := m.selfGet(key)

	if err != nil {
		if !etcd.IsKeyNotFound(err) {
			return "", fmt.Errorf(
				"error while getting variable key=%s for machine=%s: %s",
				key, m.mac, err)
		}

		// Key was not found for the machine
		value, err := m.etcdDS.GetClusterVariable(key)
		if err != nil {
			if !etcd.IsKeyNotFound(err) {
				return "", fmt.Errorf(
					"error while getting variable key=%s for machine=%s (global check): %s",
					key, m.mac, err)

			}
			return "", nil // Not set, not for machine, nor globally
		}
		return value, nil
	}

	return value, nil
}

func (self *NsqLookupdEtcdMgr) GetTopicInfo(topic string, partition int) (*TopicPartitionMetaInfo, error) {
	var topicInfo TopicPartitionMetaInfo
	self.tmiMutex.Lock()
	metaInfo, ok := self.topicMetaMap[topic]
	if !ok {
		rsp, err := self.client.Get(self.createTopicMetaPath(topic), false, false)
		if err != nil {

			self.tmiMutex.Unlock()

			if client.IsKeyNotFound(err) {
				atomic.StoreInt32(&self.ifTopicChanged, 1)
				return nil, ErrKeyNotFound
			}
			return nil, err
		}
		var mInfo TopicMetaInfo
		err = json.Unmarshal([]byte(rsp.Node.Value), &mInfo)
		if err != nil {
			self.tmiMutex.Unlock()

			return nil, err
		}
		self.topicMetaMap[topic] = &mInfo
		metaInfo = &mInfo
	}
	self.tmiMutex.Unlock()

	topicInfo.TopicMetaInfo = *metaInfo
	rsp, err := self.client.Get(self.createTopicReplicaInfoPath(topic, partition), false, false)
	if err != nil {
		if client.IsKeyNotFound(err) {
			atomic.StoreInt32(&self.ifTopicChanged, 1)
			return nil, ErrKeyNotFound
		}
		return nil, err
	}
	var rInfo TopicPartitionReplicaInfo
	if err = json.Unmarshal([]byte(rsp.Node.Value), &rInfo); err != nil {
		return nil, err
	}
	rInfo.Epoch = EpochType(rsp.Node.ModifiedIndex)
	topicInfo.TopicPartitionReplicaInfo = rInfo
	topicInfo.Name = topic
	topicInfo.Partition = partition

	return &topicInfo, nil
}

func (self *NsqdEtcdMgr) ReleaseTopicLeader(topic string, partition int, session *TopicLeaderSession) error {
	self.Lock()
	defer self.Unlock()

	topicKey := self.createTopicLeaderPath(topic, partition)
	valueB, err := json.Marshal(session)
	if err != nil {
		return err
	}

	_, err = self.client.CompareAndDelete(topicKey, string(valueB), 0)
	if err != nil {
		if !client.IsKeyNotFound(err) {
			coordLog.Errorf("try release topic leader session [%s] error: %v, orig: %v", topicKey, err, session)
			// since the topic leader session type is changed, we need do the compatible check
			rsp, innErr := self.client.Get(topicKey, false, false)
			if innErr != nil {
			} else {
				var old TopicLeaderSessionOld
				json.Unmarshal([]byte(rsp.Node.Value), &old)
				if old.IsEqual(session) {
					_, err = self.client.CompareAndDelete(topicKey, rsp.Node.Value, 0)
				}
			}
		}
	}
	if err == nil {
		coordLog.Infof("try release topic leader session [%s] success: %v", topicKey, session)
	}
	return err
}

// Machine creates a record for the associated mac if needed
// and asked for, and returns a Machine with the stored values.
// If createIfNeeded is true, and there is no machine associated to
// this mac, the machine will be created, stored, and returned.
// In this case, if createWithIP is empty, the IP will be assigned
// automatically, otherwise the given will be used. An error will be
// raised if createWithIP is currently assigned to another mac. Also
// the Type will be automatically set to MTNormal if createWithIP is
// nil, otherwise to MTStatic.
// If createIfNeeded is false, the createWithIP is expected to be nil.
// Note: if the machine exists, createWithIP is ignored. It's possible
// for the returned Machine to have an IP different from createWithIP.
func (m *etcdMachineInterface) Machine(createIfNeeded bool,
	createWithIP net.IP) (Machine, error) {
	var machine Machine

	if !createIfNeeded && (createWithIP != nil) {
		return machine, errors.New(
			"if createIfNeeded is false, the createWithIP is expected to be nil")
	}

	resp, err := m.selfGet("_machine")
	if err != nil {
		errorIsKeyNotFound := etcd.IsKeyNotFound(err)

		if !(errorIsKeyNotFound && createIfNeeded) {
			return machine, fmt.Errorf("error while retrieving _machine: %s", err)
		}

		machine := Machine{
			IP:        createWithIP, // to be assigned automatically
			FirstSeen: time.Now().Unix(),
		}
		err := m.store(&machine)
		if err != nil {
			return machine, fmt.Errorf("error while storing _machine: %s", err)
		}
		return machine, nil
	}
	json.Unmarshal([]byte(resp), &machine)
	return machine, nil
}

func (self *NsqdEtcdMgr) GetTopicInfo(topic string, partition int) (*TopicPartitionMetaInfo, error) {
	var topicInfo TopicPartitionMetaInfo
	rsp, err := self.client.Get(self.createTopicMetaPath(topic), false, false)
	if err != nil {
		if client.IsKeyNotFound(err) {
			return nil, ErrKeyNotFound
		}
		return nil, err
	}
	var mInfo TopicMetaInfo
	err = json.Unmarshal([]byte(rsp.Node.Value), &mInfo)
	if err != nil {
		return nil, err
	}
	topicInfo.TopicMetaInfo = mInfo

	rsp, err = self.client.Get(self.createTopicReplicaInfoPath(topic, partition), false, false)
	if err != nil {
		return nil, err
	}
	var rInfo TopicPartitionReplicaInfo
	if err = json.Unmarshal([]byte(rsp.Node.Value), &rInfo); err != nil {
		return nil, err
	}
	rInfo.Epoch = EpochType(rsp.Node.ModifiedIndex)
	topicInfo.TopicPartitionReplicaInfo = rInfo
	topicInfo.Name = topic
	topicInfo.Partition = partition

	return &topicInfo, nil
}

// MkDir creates an empty etcd directory
func (etcdClient *SimpleEtcdClient) MkDir(directory string) error {
	api := client.NewKeysAPI(etcdClient.etcd)
	results, err := api.Get(context.Background(), directory, nil)

	if err != nil && !client.IsKeyNotFound(err) {
		return err
	}

	if err != nil && client.IsKeyNotFound(err) {
		_, err = api.Set(context.Background(), directory, "", &client.SetOptions{Dir: true, PrevExist: client.PrevIgnore})
		return err
	}

	if !results.Node.Dir {
		return fmt.Errorf("Refusing to overwrite key/value with a directory: %v", directory)
	}
	return nil
}

func RepoUserExistsByEmail(email string) (bool, string, error) {
	userId, err := RepoGetUserIdByEmail(email)
	if err != nil {
		if client.IsKeyNotFound(err) != true {
			return false, userId, err
		}
		return false, userId, nil
	}
	return true, userId, nil
}

// Del deletes a key from Etcd
func (etcdClient *SimpleEtcdClient) Del(key string) error {
	api := client.NewKeysAPI(etcdClient.etcd)
	_, err := api.Delete(context.Background(), key, nil)
	if err != nil {
		if client.IsKeyNotFound(err) {
			return nil
		}
	}
	return err
}

// DelDir deletes a dir from Etcd
func (etcdClient *SimpleEtcdClient) DelDir(key string) error {
	api := client.NewKeysAPI(etcdClient.etcd)
	_, err := api.Delete(context.Background(), key, &client.DeleteOptions{Dir: true, Recursive: true})
	if err != nil {
		if client.IsKeyNotFound(err) {
			return nil
		}
	}
	return err
}

func RepoDomainExistsByName(name string) (bool, Domain, error) {
	domain, err := RepoGetDomainByName(name)
	if err != nil {
		if client.IsKeyNotFound(err) != true {
			return false, domain, err
		}
		return false, domain, nil
	}
	return true, domain, nil
}

func (self *NsqLookupdEtcdMgr) GetClusterEpoch() (EpochType, error) {
	rsp, err := self.client.Get(self.clusterPath, false, false)
	if err != nil {
		if client.IsKeyNotFound(err) {
			return 0, ErrKeyNotFound
		}
		return 0, err
	}

	return EpochType(rsp.Node.ModifiedIndex), nil
}

// Get gets a value in Etcd
func (etcdClient *SimpleEtcdClient) Get(key string) (string, error) {
	api := client.NewKeysAPI(etcdClient.etcd)
	response, err := api.Get(context.Background(), key, nil)
	if err != nil {
		if client.IsKeyNotFound(err) {
			return "", nil
		}
		return "", err
	}
	return response.Node.Value, nil
}

// LsRecursive returns all the keys available in the directory, recursively
func (etcdClient *SimpleEtcdClient) LsRecursive(directory string) ([]string, error) {
	api := client.NewKeysAPI(etcdClient.etcd)
	options := &client.GetOptions{Sort: true, Recursive: true}
	response, err := api.Get(context.Background(), directory, options)

	if err != nil {
		if client.IsKeyNotFound(err) {
			return make([]string, 0), nil
		}
		return make([]string, 0), err
	}

	return nodesToStringSlice(response.Node.Nodes), nil
}

func (self *NsqdEtcdMgr) GetTopicLeaderSession(topic string, partition int) (*TopicLeaderSession, error) {
	rsp, err := self.client.Get(self.createTopicLeaderPath(topic, partition), false, false)
	if err != nil {
		if client.IsKeyNotFound(err) {
			return nil, ErrKeyNotFound
		}
		return nil, err
	}
	var topicLeaderSession TopicLeaderSession
	if err = json.Unmarshal([]byte(rsp.Node.Value), &topicLeaderSession); err != nil {
		return nil, err
	}
	return &topicLeaderSession, nil
}

func (self *NsqLookupdEtcdMgr) GetTopicMetaInfo(topic string) (TopicMetaInfo, EpochType, error) {
	var metaInfo TopicMetaInfo
	rsp, err := self.client.Get(self.createTopicMetaPath(topic), false, false)
	if err != nil {
		if client.IsKeyNotFound(err) {
			return metaInfo, 0, ErrKeyNotFound
		}
		return metaInfo, 0, err
	}
	err = json.Unmarshal([]byte(rsp.Node.Value), &metaInfo)
	if err != nil {
		return metaInfo, 0, err
	}
	epoch := EpochType(rsp.Node.ModifiedIndex)
	return metaInfo, epoch, nil
}

func RepoCreateUniqueId(tries int) (string, error) {
	var id string
	try := 0
	for try < tries {
		try++
		uuid, err := uuid.NewV4()
		if err != nil {
			continue
		}
		id = uuid.String()
		_, err = RepoGetUserById(id)
		if err != nil && client.IsKeyNotFound(err) {
			return id, nil
		}
	}
	return id, errors.New("Unable to Create UUID!")
}

func (rrsets ResourceRecordSets) Get(name string) (dnsprovider.ResourceRecordSet, error) {
	getOpts := &etcdc.GetOptions{
		Recursive: true,
	}
	etcdPathPrefix := rrsets.zone.zones.intf.etcdPathPrefix
	response, err := rrsets.zone.zones.intf.etcdKeysAPI.Get(context.Background(), dnsmsg.Path(name, etcdPathPrefix), getOpts)
	if err != nil {
		if etcdc.IsKeyNotFound(err) {
			glog.V(2).Infof("Subdomain %q does not exist", name)
			return nil, nil
		}
		return nil, fmt.Errorf("Failed to get service from etcd, err: %v", err)
	}
	if emptyResponse(response) {
		glog.V(2).Infof("Subdomain %q does not exist in etcd", name)
		return nil, nil
	}

	rrset := ResourceRecordSet{name: name, rrdatas: []string{}, rrsets: &rrsets}
	found := false
	for _, node := range response.Node.Nodes {
		found = true
		service := dnsmsg.Service{}
		err = json.Unmarshal([]byte(node.Value), &service)
		if err != nil {
			return nil, fmt.Errorf("Failed to unmarshall json data, err: %v", err)
		}

		// assuming all rrdatas in a rrset will have same type
		ip := net.ParseIP(service.Host)
		switch {
		case ip == nil:
			rrset.rrsType = rrstype.CNAME
		case ip.To4() != nil:
			rrset.rrsType = rrstype.A
		}
		rrset.rrdatas = append(rrset.rrdatas, service.Host)
		rrset.ttl = int64(service.TTL)
	}

	if !found {
		return nil, nil
	}

	return rrset, nil
}

func (self *NsqLookupdEtcdMgr) GetAllLookupdNodes() ([]NsqLookupdNodeInfo, error) {
	rsp, err := self.client.Get(self.lookupdRootPath, false, false)
	if err != nil {
		if client.IsKeyNotFound(err) {
			return nil, ErrKeyNotFound
		}
		return nil, err
	}
	lookupdNodeList := make([]NsqLookupdNodeInfo, 0)
	for _, node := range rsp.Node.Nodes {
		var nodeInfo NsqLookupdNodeInfo
		if err = json.Unmarshal([]byte(node.Value), &nodeInfo); err != nil {
			continue
		}
		lookupdNodeList = append(lookupdNodeList, nodeInfo)
	}
	return lookupdNodeList, nil
}

func (self *NsqLookupdEtcdMgr) scanTopics() ([]TopicPartitionMetaInfo, error) {
	rsp, err := self.client.Get(self.topicRoot, true, true)
	if err != nil {
		atomic.StoreInt32(&self.ifTopicChanged, 1)
		if client.IsKeyNotFound(err) {
			return nil, ErrKeyNotFound
		}
		return nil, err
	}
	atomic.StoreInt32(&self.ifTopicChanged, 0)

	topicMetaMap := make(map[string]TopicMetaInfo)
	topicReplicasMap := make(map[string]map[string]TopicPartitionReplicaInfo)
	self.processTopicNode(rsp.Node.Nodes, topicMetaMap, topicReplicasMap)

	topicMetaInfos := make([]TopicPartitionMetaInfo, 0)
	for k, v := range topicReplicasMap {
		topicMeta, ok := topicMetaMap[k]
		if !ok {
			continue
		}
		for k2, v2 := range v {
			partition, err := strconv.Atoi(k2)
			if err != nil {
				continue
			}
			var topicInfo TopicPartitionMetaInfo
			topicInfo.Name = k
			topicInfo.Partition = partition
			topicInfo.TopicMetaInfo = topicMeta
			topicInfo.TopicPartitionReplicaInfo = v2
			topicMetaInfos = append(topicMetaInfos, topicInfo)
		}
	}

	self.tmisMutex.Lock()
	self.topicMetaInfos = topicMetaInfos
	self.tmisMutex.Unlock()

	return topicMetaInfos, nil
}

func (self *NsqLookupdEtcdMgr) DeleteTopic(topic string, partition int) error {
	_, err := self.client.Delete(self.createTopicPartitionPath(topic, partition), true)
	if err != nil {
		if !client.IsKeyNotFound(err) {
			return err
		}
	}
	// stop watch topic leader and delete
	topicLeaderSession := self.createTopicLeaderSessionPath(topic, partition)

	self.wtliMutex.Lock()
	defer self.wtliMutex.Unlock()
	v, ok := self.watchTopicLeaderChanMap[topicLeaderSession]
	if ok {
		close(v.watchStopCh)
		<-v.stoppedCh
		delete(self.watchTopicLeaderChanMap, topicLeaderSession)
	}

	return nil
}