// List child nodes of a given directory
func (s *Zookeeper) List(directory string) ([]*store.KVPair, error) {
	keys, stat, err := s.client.Children(store.Normalize(directory))
	if err != nil {
		if err == zk.ErrNoNode {
			return nil, store.ErrKeyNotFound
		}
		return nil, err
	}

	kv := []*store.KVPair{}

	// FIXME Costly Get request for each child key..
	for _, key := range keys {
		pair, err := s.Get(directory + store.Normalize(key))
		if err != nil {
			// If node is not found: List is out of date, retry
			if err == zk.ErrNoNode {
				return s.List(directory)
			}
			return nil, err
		}

		kv = append(kv, &store.KVPair{
			Key:       key,
			Value:     []byte(pair.Value),
			LastIndex: uint64(stat.Version),
		})
	}

	return kv, nil
}

// Get the value at "key", returns the last modified index
// to use in conjunction to Atomic calls
func (s *Zookeeper) Get(key string) (pair *store.KVPair, err error) {
	resp, meta, err := s.client.Get(store.Normalize(key))
	if err != nil {
		return nil, err
	}

	// If resp is nil, the key does not exist
	if resp == nil {
		return nil, store.ErrKeyNotFound
	}

	// FIXME handle very rare cases where Get returns the
	// SOH control character instead of the actual value
	if string(resp) == SOH {
		return s.Get(store.Normalize(key))
	}

	pair = &store.KVPair{
		Key:       key,
		Value:     resp,
		LastIndex: uint64(meta.Version),
	}

	return pair, nil
}

// Exists checks if the key exists inside the store
func (s *Zookeeper) Exists(key string) (bool, error) {
	exists, _, err := s.client.Exists(store.Normalize(key))
	if err != nil {
		return false, err
	}
	return exists, nil
}

// DeleteTree deletes a range of keys under a given directory
func (s *Etcd) DeleteTree(directory string) error {
	_, err := s.client.Delete(store.Normalize(directory), true)
	if isKeyNotFoundError(err) {
		return store.ErrKeyNotFound
	}
	return err
}

// Delete a value at "key"
func (s *Etcd) Delete(key string) error {
	_, err := s.client.Delete(store.Normalize(key), false)
	if isKeyNotFoundError(err) {
		return store.ErrKeyNotFound
	}
	return err
}

// AtomicPut put a value at "key" if the key has not been
// modified in the meantime, throws an error if this is the case
func (s *Zookeeper) AtomicPut(key string, value []byte, previous *store.KVPair, _ *store.WriteOptions) (bool, *store.KVPair, error) {

	var lastIndex uint64
	if previous != nil {
		meta, err := s.client.Set(store.Normalize(key), value, int32(previous.LastIndex))
		if err != nil {
			// Compare Failed
			if err == zk.ErrBadVersion {
				return false, nil, store.ErrKeyModified
			}
			return false, nil, err
		}
		lastIndex = uint64(meta.Version)
	} else {
		// Interpret previous == nil as create operation.
		_, err := s.client.Create(store.Normalize(key), value, 0, zk.WorldACL(zk.PermAll))
		if err != nil {
			// Zookeeper will complain if the directory doesn't exist.
			if err == zk.ErrNoNode {
				// Create the directory
				parts := store.SplitKey(key)
				parts = parts[:len(parts)-1]
				if err = s.createFullPath(parts, false); err != nil {
					// Failed to create the directory.
					return false, nil, err
				}
				if _, err := s.client.Create(store.Normalize(key), value, 0, zk.WorldACL(zk.PermAll)); err != nil {
					return false, nil, err
				}

			} else {
				// Unhandled error
				return false, nil, err
			}
		}
		lastIndex = 0 // Newly created nodes have version 0.
	}

	pair := &store.KVPair{
		Key:       key,
		Value:     value,
		LastIndex: lastIndex,
	}

	return true, pair, nil
}

// AtomicPut put a value at "key" if the key has not been
// modified in the meantime, throws an error if this is the case
func (s *Etcd) AtomicPut(key string, value []byte, previous *store.KVPair, options *store.WriteOptions) (bool, *store.KVPair, error) {

	var meta *etcd.Response
	var err error
	if previous != nil {
		meta, err = s.client.CompareAndSwap(store.Normalize(key), string(value), 0, "", previous.LastIndex)
	} else {
		// Interpret previous == nil as Atomic Create
		meta, err = s.client.Create(store.Normalize(key), string(value), 0)
		if etcdError, ok := err.(*etcd.EtcdError); ok {

			// Directory doesn't exist.
			if etcdError.ErrorCode == 104 {
				// Remove the last element (the actual key)
				// and create the full directory path
				err = s.createDirectory(store.GetDirectory(key))
				if err != nil {
					return false, nil, err
				}

				// Now that the directory is created, create the key
				if _, err := s.client.Create(key, string(value), 0); err != nil {
					return false, nil, err
				}
			}
		}
	}
	if err != nil {
		if etcdError, ok := err.(*etcd.EtcdError); ok {
			// Compare Failed
			if etcdError.ErrorCode == 101 {
				return false, nil, store.ErrKeyModified
			}
		}
		return false, nil, err
	}

	updated := &store.KVPair{
		Key:       key,
		Value:     value,
		LastIndex: meta.Node.ModifiedIndex,
	}

	return true, updated, nil
}

// NewLock returns a handle to a lock struct which can
// be used to provide mutual exclusion on a key
func (s *Zookeeper) NewLock(key string, options *store.LockOptions) (lock store.Locker, err error) {
	value := []byte("")

	// Apply options
	if options != nil {
		if options.Value != nil {
			value = options.Value
		}
	}

	lock = &zookeeperLock{
		client: s.client,
		key:    store.Normalize(key),
		value:  value,
		lock:   zk.NewLock(s.client, store.Normalize(key), zk.WorldACL(zk.PermAll)),
	}

	return lock, err
}

// Lock attempts to acquire the lock and blocks while
// doing so. It returns a channel that is closed if our
// lock is lost or if an error occurs
func (l *etcdLock) Lock(stopChan chan struct{}) (<-chan struct{}, error) {

	key := store.Normalize(l.key)

	// Lock holder channel
	lockHeld := make(chan struct{})
	stopLocking := l.stopRenew

	var lastIndex uint64

	for {
		resp, err := l.client.Create(key, l.value, l.ttl)
		if err != nil {
			if etcdError, ok := err.(*etcd.EtcdError); ok {
				// Key already exists
				if etcdError.ErrorCode != 105 {
					lastIndex = ^uint64(0)
				}
			}
		} else {
			lastIndex = resp.Node.ModifiedIndex
		}

		l.last, err = l.client.CompareAndSwap(key, l.value, l.ttl, "", lastIndex)

		if err == nil {
			// Leader section
			l.stopLock = stopLocking
			go l.holdLock(key, lockHeld, stopLocking)
			break
		} else {
			// Seeker section
			chW := make(chan *etcd.Response)
			chWStop := make(chan bool)
			free := make(chan bool)

			go l.waitLock(key, chW, chWStop, free)

			// Wait for the key to be available or for
			// a signal to stop trying to lock the key
			select {
			case _ = <-free:
				break
			case _ = <-stopChan:
				return nil, nil
			}

			// Delete or Expire event occured
			// Retry
		}
	}

	return lockHeld, nil
}

// Unlock the "key". Calling unlock while
// not holding the lock will throw an error
func (l *etcdLock) Unlock() error {
	if l.stopLock != nil {
		l.stopLock <- struct{}{}
	}
	if l.last != nil {
		_, err := l.client.CompareAndDelete(store.Normalize(l.key), l.value, l.last.Node.ModifiedIndex)
		if err != nil {
			return err
		}
	}
	return nil
}

// createDirectory creates the entire path for a directory
// that does not exist
func (s *Etcd) createDirectory(path string) error {
	if _, err := s.client.CreateDir(store.Normalize(path), 10); err != nil {
		if etcdError, ok := err.(*etcd.EtcdError); ok {
			// Skip key already exists
			if etcdError.ErrorCode != 105 {
				return err
			}
		} else {
			return err
		}
	}
	return nil
}

// AtomicDelete deletes a value at "key" if the key
// has not been modified in the meantime, throws an
// error if this is the case
func (s *Zookeeper) AtomicDelete(key string, previous *store.KVPair) (bool, error) {
	if previous == nil {
		return false, store.ErrPreviousNotSpecified
	}

	err := s.client.Delete(store.Normalize(key), int32(previous.LastIndex))
	if err != nil {
		if err == zk.ErrBadVersion {
			return false, store.ErrKeyModified
		}
		return false, err
	}
	return true, nil
}

// List child nodes of a given directory
func (s *Zookeeper) List(directory string) ([]*store.KVPair, error) {
	keys, stat, err := s.client.Children(store.Normalize(directory))
	if err != nil {
		return nil, err
	}

	kv := []*store.KVPair{}

	// FIXME Costly Get request for each child key..
	for _, key := range keys {
		pair, err := s.Get(directory + store.Normalize(key))
		if err != nil {
			return nil, err
		}

		kv = append(kv, &store.KVPair{
			Key:       key,
			Value:     []byte(pair.Value),
			LastIndex: uint64(stat.Version),
		})
	}

	return kv, nil
}

// List child nodes of a given directory
func (s *Etcd) List(directory string) ([]*store.KVPair, error) {
	resp, err := s.client.Get(store.Normalize(directory), true, true)
	if err != nil {
		return nil, err
	}
	kv := []*store.KVPair{}
	for _, n := range resp.Node.Nodes {
		key := strings.TrimLeft(n.Key, "/")
		kv = append(kv, &store.KVPair{
			Key:       key,
			Value:     []byte(n.Value),
			LastIndex: n.ModifiedIndex,
		})
	}
	return kv, nil
}

// Get the value at "key", returns the last modified index
// to use in conjunction to Atomic calls
func (s *Etcd) Get(key string) (pair *store.KVPair, err error) {
	result, err := s.client.Get(store.Normalize(key), false, false)
	if err != nil {
		if isKeyNotFoundError(err) {
			return nil, store.ErrKeyNotFound
		}
		return nil, err
	}

	pair = &store.KVPair{
		Key:       key,
		Value:     []byte(result.Node.Value),
		LastIndex: result.Node.ModifiedIndex,
	}

	return pair, nil
}

// DeleteTree deletes a range of keys under a given directory
func (s *Zookeeper) DeleteTree(directory string) error {
	pairs, err := s.List(directory)
	if err != nil {
		return err
	}

	var reqs []interface{}

	for _, pair := range pairs {
		reqs = append(reqs, &zk.DeleteRequest{
			Path:    store.Normalize(directory + "/" + pair.Key),
			Version: -1,
		})
	}

	_, err = s.client.Multi(reqs...)
	return err
}

// AtomicDelete deletes a value at "key" if the key
// has not been modified in the meantime, throws an
// error if this is the case
func (s *Etcd) AtomicDelete(key string, previous *store.KVPair) (bool, error) {
	if previous == nil {
		return false, store.ErrPreviousNotSpecified
	}

	_, err := s.client.CompareAndDelete(store.Normalize(key), "", previous.LastIndex)
	if err != nil {
		if etcdError, ok := err.(*etcd.EtcdError); ok {
			// Compare failed
			if etcdError.ErrorCode == 101 {
				return false, store.ErrKeyModified
			}
		}
		return false, err
	}

	return true, nil
}

// Lock attempts to acquire the lock and blocks while
// doing so. It returns a channel that is closed if our
// lock is lost or if an error occurs
func (l *etcdLock) Lock() (<-chan struct{}, error) {

	key := store.Normalize(l.key)

	// Lock holder channels
	lockHeld := make(chan struct{})
	stopLocking := make(chan struct{})

	var lastIndex uint64

	for {
		resp, err := l.client.Create(key, l.value, l.ttl)
		if err != nil {
			if etcdError, ok := err.(*etcd.EtcdError); ok {
				// Key already exists
				if etcdError.ErrorCode != 105 {
					lastIndex = ^uint64(0)
				}
			}
		} else {
			lastIndex = resp.Node.ModifiedIndex
		}

		l.last, err = l.client.CompareAndSwap(key, l.value, l.ttl, "", lastIndex)

		if err == nil {
			// Leader section
			l.stopLock = stopLocking
			go l.holdLock(key, lockHeld, stopLocking)
			break
		} else {
			// Seeker section
			chW := make(chan *etcd.Response)
			chWStop := make(chan bool)
			l.waitLock(key, chW, chWStop)

			// Delete or Expire event occured
			// Retry
		}
	}

	return lockHeld, nil
}

// Watch for changes on a "key"
// It returns a channel that will receive changes or pass
// on errors. Upon creation, the current value will first
// be sent to the channel. Providing a non-nil stopCh can
// be used to stop watching.
func (s *Etcd) Watch(key string, stopCh <-chan struct{}) (<-chan *store.KVPair, error) {
	// Start an etcd watch.
	// Note: etcd will send the current value through the channel.
	etcdWatchCh := make(chan *etcd.Response)
	etcdStopCh := make(chan bool)
	go s.client.Watch(store.Normalize(key), 0, false, etcdWatchCh, etcdStopCh)

	// Adapter goroutine: The goal here is to convert whatever
	// format etcd is using into our interface.
	watchCh := make(chan *store.KVPair)
	go func() {
		defer close(watchCh)

		// Get the current value
		current, err := s.Get(key)
		if err != nil {
			return
		}

		// Push the current value through the channel.
		watchCh <- current

		for {
			select {
			case result := <-etcdWatchCh:
				if result == nil || result.Node == nil {
					// Something went wrong, exit
					// No need to stop the chan as the watch already ended
					return
				}
				watchCh <- &store.KVPair{
					Key:       key,
					Value:     []byte(result.Node.Value),
					LastIndex: result.Node.ModifiedIndex,
				}
			case <-stopCh:
				etcdStopCh <- true
				return
			}
		}
	}()
	return watchCh, nil
}

// Put a value at "key"
func (s *Zookeeper) Put(key string, value []byte, opts *store.WriteOptions) error {
	fkey := store.Normalize(key)

	exists, err := s.Exists(key)
	if err != nil {
		return err
	}

	if !exists {
		if opts != nil && opts.Ephemeral {
			s.createFullPath(store.SplitKey(key), opts.Ephemeral)
		} else {
			s.createFullPath(store.SplitKey(key), false)
		}
	}

	_, err = s.client.Set(fkey, value, -1)
	return err
}