// Sets the value for a given key.
func SetKeyHandler(w http.ResponseWriter, req *http.Request, s Server) error {
	vars := mux.Vars(req)
	key := "/" + vars["key"]

	req.ParseForm()

	// Parse non-blank value.
	value := req.Form.Get("value")
	if len(value) == 0 {
		return etcdErr.NewError(200, "Set", s.Store().Index())
	}

	// Convert time-to-live to an expiration time.
	expireTime, err := store.TTL(req.Form.Get("ttl"))
	if err != nil {
		return etcdErr.NewError(202, "Set", s.Store().Index())
	}

	// If the "prevValue" is specified then test-and-set. Otherwise create a new key.
	var c raft.Command
	if prevValueArr, ok := req.Form["prevValue"]; ok {
		if len(prevValueArr[0]) > 0 {
			// test against previous value
			c = s.Store().CommandFactory().CreateCompareAndSwapCommand(key, value, prevValueArr[0], 0, expireTime)
		} else {
			// test against existence
			c = s.Store().CommandFactory().CreateCreateCommand(key, false, value, expireTime, false)
		}

	} else {
		c = s.Store().CommandFactory().CreateSetCommand(key, false, value, expireTime)
	}

	return s.Dispatch(c, w, req)
}

// InternalGet gets the node of the given nodePath.
func (s *store) internalGet(nodePath string) (*node, *etcdErr.Error) {
	nodePath = path.Clean(path.Join("/", nodePath))

	walkFunc := func(parent *node, name string) (*node, *etcdErr.Error) {

		if !parent.IsDir() {
			err := etcdErr.NewError(etcdErr.EcodeNotDir, parent.Path, s.CurrentIndex)
			return nil, err
		}

		child, ok := parent.Children[name]
		if ok {
			return child, nil
		}

		return nil, etcdErr.NewError(etcdErr.EcodeKeyNotFound, path.Join(parent.Path, name), s.CurrentIndex)
	}

	f, err := s.walk(nodePath, walkFunc)

	if err != nil {
		return nil, err
	}
	return f, nil
}

// Watches a given key prefix for changes.
func WatchKeyHandler(w http.ResponseWriter, req *http.Request, s Server) error {
	var err error
	vars := mux.Vars(req)
	key := "/" + vars["key"]

	// Create a command to watch from a given index (default 0).
	var sinceIndex uint64 = 0
	if req.Method == "POST" {
		sinceIndex, err = strconv.ParseUint(string(req.FormValue("index")), 10, 64)
		if err != nil {
			return etcdErr.NewError(203, "Watch From Index", s.Store().Index())
		}
	}

	// Start the watcher on the store.
	watcher, err := s.Store().Watch(key, false, false, sinceIndex)
	if err != nil {
		return etcdErr.NewError(500, key, s.Store().Index())
	}
	event := <-watcher.EventChan

	// Convert event to a response and write to client.
	b, _ := json.Marshal(event.Response(s.Store().Index()))
	w.WriteHeader(http.StatusOK)
	w.Write(b)

	return nil
}

func (e EtcdConnector) Set(key string, dir bool, value string, ttl string, w http.ResponseWriter, req *http.Request) error {
	ps := e.etcd.PeerServer
	registry := e.etcd.Registry
	if ps.RaftServer().State() == raft.Leader {
		expireTime, err := store.TTL(ttl)
		if err != nil {
			return etcdErr.NewError(etcdErr.EcodeTTLNaN, "Create", e.etcd.EtcdServer.Store().Index())
		}
		c := e.etcd.EtcdServer.Store().CommandFactory().CreateSetCommand(key, dir, value, expireTime)
		result, err := ps.RaftServer().Do(c)
		if err != nil {
			return err
		}

		if result == nil {
			return etcdErr.NewError(300, "Empty result from raft", e.etcd.EtcdServer.Store().Index())
		}

		if w != nil {
			w.WriteHeader(http.StatusOK)
		}

		return nil
	} else {
		leader := ps.RaftServer().Leader()

		// No leader available.
		if leader == "" {
			return etcdErr.NewError(300, "", e.etcd.EtcdServer.Store().Index())
		}

		leaderUrl, _ := registry.ClientURL(leader)

		client := http.DefaultClient
		v := "?"
		if !dir {
			v += "value=" + value + "&ttl=" + ttl
		} else {
			v += "dir=" + strconv.FormatBool(dir) + "&ttl=" + ttl
		}

		req, _ := http.NewRequest("PUT", leaderUrl+"/v2/keys"+key+v, nil)
		resp, err := client.Do(req)
		resp.Body.Close()

		if err != nil {
			log.Println(err)
			return err
		} else {
			if w != nil {
				w.WriteHeader(http.StatusOK)
			}

			return nil
		}
	}

	return nil
}

// acquireHandler attempts to acquire a lock on the given key.
// The "key" parameter specifies the resource to lock.
// The "value" parameter specifies a value to associate with the lock.
// The "ttl" parameter specifies how long the lock will persist for.
// The "timeout" parameter specifies how long the request should wait for the lock.
func (h *handler) acquireHandler(w http.ResponseWriter, req *http.Request) error {
	h.client.SyncCluster()

	// Setup connection watcher.
	closeNotifier, _ := w.(http.CloseNotifier)
	closeChan := closeNotifier.CloseNotify()
	stopChan := make(chan bool)

	// Parse the lock "key".
	vars := mux.Vars(req)
	keypath := path.Join(prefix, vars["key"])
	value := req.FormValue("value")

	// Parse "timeout" parameter.
	var timeout int
	var err error
	if req.FormValue("timeout") == "" {
		timeout = -1
	} else if timeout, err = strconv.Atoi(req.FormValue("timeout")); err != nil {
		return etcdErr.NewError(etcdErr.EcodeTimeoutNaN, "Acquire", 0)
	}
	timeout = timeout + 1

	// Parse TTL.
	ttl, err := strconv.Atoi(req.FormValue("ttl"))
	if err != nil {
		return etcdErr.NewError(etcdErr.EcodeTTLNaN, "Acquire", 0)
	}

	// If node exists then just watch it. Otherwise create the node and watch it.
	node, index, pos := h.findExistingNode(keypath, value)
	if index > 0 {
		if pos == 0 {
			// If lock is already acquired then update the TTL.
			h.client.Update(node.Key, node.Value, uint64(ttl))
		} else {
			// Otherwise watch until it becomes acquired (or errors).
			err = h.watch(keypath, index, nil)
		}
	} else {
		index, err = h.createNode(keypath, value, ttl, closeChan, stopChan)
	}

	// Stop all goroutines.
	close(stopChan)

	// Check for an error.
	if err != nil {
		return err
	}

	// Write response.
	w.Write([]byte(strconv.Itoa(index)))
	return nil
}

// Remove function remove the node.
func (n *node) Remove(dir, recursive bool, callback func(path string)) *etcdErr.Error {

	if n.IsDir() {
		if !dir {
			// cannot delete a directory without recursive set to true
			return etcdErr.NewError(etcdErr.EcodeNotFile, n.Path, n.store.Index())
		}

		if len(n.Children) != 0 && !recursive {
			// cannot delete a directory if it is not empty and the operation
			// is not recursive
			return etcdErr.NewError(etcdErr.EcodeDirNotEmpty, n.Path, n.store.Index())
		}
	}

	if !n.IsDir() { // key-value pair
		_, name := path.Split(n.Path)

		// find its parent and remove the node from the map
		if n.Parent != nil && n.Parent.Children[name] == n {
			delete(n.Parent.Children, name)
		}

		if callback != nil {
			callback(n.Path)
		}

		if !n.IsPermanent() {
			n.store.ttlKeyHeap.remove(n)
		}

		return nil
	}

	for _, child := range n.Children { // delete all children
		child.Remove(true, true, callback)
	}

	// delete self
	_, name := path.Split(n.Path)
	if n.Parent != nil && n.Parent.Children[name] == n {
		delete(n.Parent.Children, name)

		if callback != nil {
			callback(n.Path)
		}

		if !n.IsPermanent() {
			n.store.ttlKeyHeap.remove(n)
		}

	}

	return nil
}

func (s *store) CompareAndSwap(nodePath string, prevValue string, prevIndex uint64,
	value string, expireTime time.Time) (*Event, error) {

	nodePath = path.Clean(path.Join("/", nodePath))
	// we do not allow the user to change "/"
	if nodePath == "/" {
		return nil, etcdErr.NewError(etcdErr.EcodeRootROnly, "/", s.CurrentIndex)
	}

	s.worldLock.Lock()
	defer s.worldLock.Unlock()

	n, err := s.internalGet(nodePath)

	if err != nil {
		s.Stats.Inc(CompareAndSwapFail)
		return nil, err
	}

	if n.IsDir() { // can only compare and swap file
		s.Stats.Inc(CompareAndSwapFail)
		return nil, etcdErr.NewError(etcdErr.EcodeNotFile, nodePath, s.CurrentIndex)
	}

	// If both of the prevValue and prevIndex are given, we will test both of them.
	// Command will be executed, only if both of the tests are successful.
	if !n.Compare(prevValue, prevIndex) {
		cause := fmt.Sprintf("[%v != %v] [%v != %v]", prevValue, n.Value, prevIndex, n.ModifiedIndex)
		s.Stats.Inc(CompareAndSwapFail)
		return nil, etcdErr.NewError(etcdErr.EcodeTestFailed, cause, s.CurrentIndex)
	}

	// update etcd index
	s.CurrentIndex++

	e := newEvent(CompareAndSwap, nodePath, s.CurrentIndex, n.CreatedIndex)
	e.PrevNode = n.Repr(false, false)
	eNode := e.Node

	// if test succeed, write the value
	n.Write(value, s.CurrentIndex)
	n.UpdateTTL(expireTime)

	eNode.Value = value
	eNode.Expiration, eNode.TTL = n.ExpirationAndTTL()

	s.WatcherHub.notify(e)
	s.Stats.Inc(CompareAndSwapSuccess)
	return e, nil
}

// renewLockHandler attempts to update the TTL on an existing lock.
// Returns a 200 OK if successful. Returns non-200 on error.
func (h *handler) renewLockHandler(w http.ResponseWriter, req *http.Request) error {
	h.client.SyncCluster()

	// Read the lock path.
	vars := mux.Vars(req)
	keypath := path.Join(prefix, vars["key"])

	// Parse new TTL parameter.
	ttl, err := strconv.Atoi(req.FormValue("ttl"))
	if err != nil {
		return etcdErr.NewError(etcdErr.EcodeTTLNaN, "Renew", 0)
	}

	// Read and set defaults for index and value.
	index := req.FormValue("index")
	value := req.FormValue("value")
	if len(index) == 0 && len(value) == 0 {
		return etcdErr.NewError(etcdErr.EcodeIndexOrValueRequired, "Renew", 0)
	}

	if len(index) == 0 {
		// If index is not specified then look it up by value.
		resp, err := h.client.Get(keypath, true, true)
		if err != nil {
			return err
		}
		nodes := lockNodes{resp.Node.Nodes}
		node, _ := nodes.FindByValue(value)
		if node == nil {
			return etcdErr.NewError(etcdErr.EcodeKeyNotFound, "Renew", 0)
		}
		index = path.Base(node.Key)

	} else if len(value) == 0 {
		// If value is not specified then default it to the previous value.
		resp, err := h.client.Get(path.Join(keypath, index), true, false)
		if err != nil {
			return err
		}
		value = resp.Node.Value
	}

	// Renew the lock, if it exists.
	if _, err = h.client.Update(path.Join(keypath, index), value, uint64(ttl)); err != nil {
		return err
	}

	return nil
}

// Join a server to the cluster
func (c *JoinCommand) Apply(context raft.Context) (interface{}, error) {
	ps, _ := context.Server().Context().(*PeerServer)

	b := make([]byte, 8)
	binary.PutUvarint(b, context.CommitIndex())

	// Make sure we're not getting a cached value from the registry.
	ps.registry.Invalidate(c.Name)

	// Check if the join command is from a previous peer, who lost all its previous log.
	if _, ok := ps.registry.ClientURL(c.Name); ok {
		return b, nil
	}

	// Check peer number in the cluster
	if ps.registry.Count() == ps.Config.MaxClusterSize {
		log.Debug("Reject join request from ", c.Name)
		return []byte{0}, etcdErr.NewError(etcdErr.EcodeNoMorePeer, "", context.CommitIndex())
	}

	// Add to shared peer registry.
	ps.registry.Register(c.Name, c.RaftURL, c.EtcdURL)

	// Add peer in raft
	err := context.Server().AddPeer(c.Name, "")

	// Add peer stats
	if c.Name != ps.RaftServer().Name() {
		ps.followersStats.Followers[c.Name] = &raftFollowerStats{}
		ps.followersStats.Followers[c.Name].Latency.Minimum = 1 << 63
	}

	return b, err
}

// Read function gets the value of the node.
// If the receiver node is not a key-value pair, a "Not A File" error will be returned.
func (n *node) Read() (string, *etcdErr.Error) {
	if n.IsDir() {
		return "", etcdErr.NewError(etcdErr.EcodeNotFile, "", n.store.Index())
	}

	return n.Value, nil
}

// deleteHandler remove a given leader.
func (h *handler) deleteHandler(w http.ResponseWriter, req *http.Request) error {
	vars := mux.Vars(req)
	name := req.FormValue("name")
	if name == "" {
		return etcdErr.NewError(etcdErr.EcodeNameRequired, "Delete", 0)
	}

	// Proxy the request to the the lock service.
	u, err := url.Parse(fmt.Sprintf("%s/mod/v2/lock/%s", h.addr, vars["key"]))
	if err != nil {
		return err
	}
	q := u.Query()
	q.Set("value", name)
	u.RawQuery = q.Encode()

	r, err := http.NewRequest("DELETE", u.String(), nil)
	if err != nil {
		return err
	}

	// Read from the leader lock.
	resp, err := h.client.Do(r)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	w.WriteHeader(resp.StatusCode)
	io.Copy(w, resp.Body)

	return nil
}

// getIndexHandler retrieves the current lock index.
// The "field" parameter specifies to read either the lock "index" or lock "value".
func (h *handler) getIndexHandler(w http.ResponseWriter, req *http.Request) error {
	h.client.SyncCluster()

	vars := mux.Vars(req)
	keypath := path.Join(prefix, vars["key"])
	field := req.FormValue("field")
	if len(field) == 0 {
		field = "value"
	}

	// Read all indices.
	resp, err := h.client.Get(keypath, true, true)
	if err != nil {
		return err
	}
	nodes := lockNodes{resp.Node.Nodes}

	// Write out the requested field.
	if node := nodes.First(); node != nil {
		switch field {
		case "index":
			w.Write([]byte(path.Base(node.Key)))

		case "value":
			w.Write([]byte(node.Value))

		default:
			return etcdErr.NewError(etcdErr.EcodeInvalidField, "Get", 0)
		}
	}

	return nil
}

func (e EtcdConnector) CreateSetCommand(key string, dir bool, value string, ttl string) (raft.Command, error) {
	expireTime, err := store.TTL(ttl)
	if err != nil {
		return nil, etcdErr.NewError(etcdErr.EcodeTTLNaN, "Create", e.etcd.EtcdServer.Store().Index())
	}
	c := e.etcd.EtcdServer.Store().CommandFactory().CreateSetCommand(key, dir, value, expireTime)
	return c, nil
}

// Add function adds a node to the receiver node.
// If the receiver is not a directory, a "Not A Directory" error will be returned.
// If there is a existing node with the same name under the directory, a "Already Exist"
// error will be returned
func (n *node) Add(child *node) *etcdErr.Error {
	if !n.IsDir() {
		return etcdErr.NewError(etcdErr.EcodeNotDir, "", n.store.Index())
	}

	_, name := path.Split(child.Path)

	_, ok := n.Children[name]

	if ok {
		return etcdErr.NewError(etcdErr.EcodeNodeExist, "", n.store.Index())
	}

	n.Children[name] = child

	return nil
}

// Update updates the value/ttl of the node.
// If the node is a file, the value and the ttl can be updated.
// If the node is a directory, only the ttl can be updated.
func (s *store) Update(nodePath string, newValue string, expireTime time.Time) (*Event, error) {
	nodePath = path.Clean(path.Join("/", nodePath))
	// we do not allow the user to change "/"
	if nodePath == "/" {
		return nil, etcdErr.NewError(etcdErr.EcodeRootROnly, "/", s.CurrentIndex)
	}

	s.worldLock.Lock()
	defer s.worldLock.Unlock()

	currIndex, nextIndex := s.CurrentIndex, s.CurrentIndex+1

	n, err := s.internalGet(nodePath)

	if err != nil { // if the node does not exist, return error
		s.Stats.Inc(UpdateFail)
		return nil, err
	}

	e := newEvent(Update, nodePath, nextIndex, n.CreatedIndex)
	e.PrevNode = n.Repr(false, false)
	eNode := e.Node

	if n.IsDir() && len(newValue) != 0 {
		// if the node is a directory, we cannot update value to non-empty
		s.Stats.Inc(UpdateFail)
		return nil, etcdErr.NewError(etcdErr.EcodeNotFile, nodePath, currIndex)
	}

	n.Write(newValue, nextIndex)
	eNode.Value = newValue

	// update ttl
	n.UpdateTTL(expireTime)

	eNode.Expiration, eNode.TTL = n.ExpirationAndTTL()

	s.WatcherHub.notify(e)

	s.Stats.Inc(UpdateSuccess)

	s.CurrentIndex = nextIndex

	return e, nil
}

func handleWatch(key string, recursive, stream bool, waitIndex string, w http.ResponseWriter, s Server) error {
	// Create a command to watch from a given index (default 0).
	var sinceIndex uint64 = 0
	var err error

	if waitIndex != "" {
		sinceIndex, err = strconv.ParseUint(waitIndex, 10, 64)
		if err != nil {
			return etcdErr.NewError(etcdErr.EcodeIndexNaN, "Watch From Index", s.Store().Index())
		}
	}

	watcher, err := s.Store().Watch(key, recursive, stream, sinceIndex)
	if err != nil {
		return err
	}

	cn, _ := w.(http.CloseNotifier)
	closeChan := cn.CloseNotify()

	writeHeaders(w, s)

	if stream {
		// watcher hub will not help to remove stream watcher
		// so we need to remove here
		defer watcher.Remove()
		for {
			select {
			case <-closeChan:
				return nil
			case event, ok := <-watcher.EventChan:
				if !ok {
					// If the channel is closed this may be an indication of
					// that notifications are much more than we are able to
					// send to the client in time. Then we simply end streaming.
					return nil
				}

				b, _ := json.Marshal(event)
				_, err := w.Write(b)
				if err != nil {
					return nil
				}
				w.(http.Flusher).Flush()
			}
		}
	}

	select {
	case <-closeChan:
		watcher.Remove()
	case event := <-watcher.EventChan:
		b, _ := json.Marshal(event)
		w.Write(b)
	}
	return nil
}

func UpdateHandler(w http.ResponseWriter, req *http.Request, s Server, key, value string, expireTime time.Time) error {
	// Update should give at least one option
	if value == "" && expireTime.Sub(store.Permanent) == 0 {
		return etcdErr.NewError(etcdErr.EcodeValueOrTTLRequired, "Update", s.Store().Index())
	}

	c := s.Store().CommandFactory().CreateUpdateCommand(key, value, expireTime)
	return s.Dispatch(c, w, req)
}

func (s *store) CompareAndDelete(nodePath string, prevValue string, prevIndex uint64) (*Event, error) {
	nodePath = path.Clean(path.Join("/", nodePath))

	s.worldLock.Lock()
	defer s.worldLock.Unlock()

	n, err := s.internalGet(nodePath)

	if err != nil { // if the node does not exist, return error
		s.Stats.Inc(CompareAndDeleteFail)
		return nil, err
	}

	if n.IsDir() { // can only compare and delete file
		s.Stats.Inc(CompareAndSwapFail)
		return nil, etcdErr.NewError(etcdErr.EcodeNotFile, nodePath, s.CurrentIndex)
	}

	// If both of the prevValue and prevIndex are given, we will test both of them.
	// Command will be executed, only if both of the tests are successful.
	if !n.Compare(prevValue, prevIndex) {
		cause := fmt.Sprintf("[%v != %v] [%v != %v]", prevValue, n.Value, prevIndex, n.ModifiedIndex)
		s.Stats.Inc(CompareAndDeleteFail)
		return nil, etcdErr.NewError(etcdErr.EcodeTestFailed, cause, s.CurrentIndex)
	}

	// update etcd index
	s.CurrentIndex++

	e := newEvent(CompareAndDelete, nodePath, s.CurrentIndex, n.CreatedIndex)
	e.PrevNode = n.Repr(false, false)

	callback := func(path string) { // notify function
		// notify the watchers with deleted set true
		s.WatcherHub.notifyWatchers(e, path, true)
	}

	// delete a key-value pair, no error should happen
	n.Remove(false, false, callback)

	s.WatcherHub.notify(e)
	s.Stats.Inc(CompareAndDeleteSuccess)
	return e, nil
}

// Handler to return the current leader's raft address
func (s *Server) GetLeaderHandler(w http.ResponseWriter, req *http.Request) error {
	leader := s.peerServer.RaftServer().Leader()
	if leader == "" {
		return etcdErr.NewError(etcdErr.EcodeLeaderElect, "", s.Store().Index())
	}
	w.WriteHeader(http.StatusOK)
	url, _ := s.registry.PeerURL(leader)
	w.Write([]byte(url))
	return nil
}

// Write function set the value of the node to the given value.
// If the receiver node is a directory, a "Not A File" error will be returned.
func (n *node) Write(value string, index uint64) *etcdErr.Error {
	if n.IsDir() {
		return etcdErr.NewError(etcdErr.EcodeNotFile, "", n.store.Index())
	}

	n.Value = value
	n.ModifiedIndex = index

	return nil
}