// ListKeys is used to list all keys with a given prefix to a seperator
func (k *KVS) ListKeys(args *structs.KeyListRequest, reply *structs.IndexedKeyList) error {
	if done, err := k.srv.forward("KVS.ListKeys", args, args, reply); done {
		return err
	}

	acl, err := k.srv.resolveToken(args.Token)
	if err != nil {
		return err
	}

	// Get the local state
	state := k.srv.fsm.State()
	return k.srv.blockingRPC(&args.QueryOptions,
		&reply.QueryMeta,
		state.QueryTables("KVSListKeys"),
		func() error {
			index, keys, err := state.KVSListKeys(args.Prefix, args.Seperator)
			reply.Index = index
			if acl != nil {
				keys = FilterKeys(acl, keys)
			}
			reply.Keys = keys
			return err
		})
}

// ListKeys is used to list all keys with a given prefix to a seperator
func (k *KVS) ListKeys(args *structs.KeyListRequest, reply *structs.IndexedKeyList) error {
	if done, err := k.srv.forward("KVS.ListKeys", args, args, reply); done {
		return err
	}

	acl, err := k.srv.resolveToken(args.Token)
	if err != nil {
		return err
	}

	// Get the local state
	state := k.srv.fsm.State()
	opts := blockingRPCOptions{
		queryOpts: &args.QueryOptions,
		queryMeta: &reply.QueryMeta,
		kvWatch:   true,
		kvPrefix:  args.Prefix,
		run: func() error {
			index, keys, err := state.KVSListKeys(args.Prefix, args.Seperator)
			reply.Index = index
			if acl != nil {
				keys = FilterKeys(acl, keys)
			}
			reply.Keys = keys
			return err

		},
	}
	return k.srv.blockingRPCOpt(&opts)
}

// KVSGetKeys handles a GET request for keys
func (s *HTTPServer) KVSGetKeys(resp http.ResponseWriter, req *http.Request, args *structs.KeyRequest) (interface{}, error) {
	// Check for a seperator, due to historic spelling error,
	// we now are forced to check for both spellings
	var sep string
	params := req.URL.Query()
	if _, ok := params["seperator"]; ok {
		sep = params.Get("seperator")
	}
	if _, ok := params["separator"]; ok {
		sep = params.Get("separator")
	}

	// Construct the args
	listArgs := structs.KeyListRequest{
		Datacenter:   args.Datacenter,
		Prefix:       args.Key,
		Seperator:    sep,
		QueryOptions: args.QueryOptions,
	}

	// Make the RPC
	var out structs.IndexedKeyList
	if err := s.agent.RPC("KVS.ListKeys", &listArgs, &out); err != nil {
		return nil, err
	}
	setMeta(resp, &out.QueryMeta)

	// Check if we get a not found. We do not generate
	// not found for the root, but just provide the empty list
	if len(out.Keys) == 0 && listArgs.Prefix != "" {
		resp.WriteHeader(404)
		return nil, nil
	}

	// Use empty list instead of null
	if out.Keys == nil {
		out.Keys = []string{}
	}
	return out.Keys, nil
}

// ListKeys is used to list all keys with a given prefix to a separator.
func (k *KVS) ListKeys(args *structs.KeyListRequest, reply *structs.IndexedKeyList) error {
	if done, err := k.srv.forward("KVS.ListKeys", args, args, reply); done {
		return err
	}

	acl, err := k.srv.resolveToken(args.Token)
	if err != nil {
		return err
	}

	// Get the local state
	state := k.srv.fsm.State()
	return k.srv.blockingRPC(
		&args.QueryOptions,
		&reply.QueryMeta,
		state.GetKVSWatch(args.Prefix),
		func() error {
			index, keys, err := state.KVSListKeys(args.Prefix, args.Seperator)
			if err != nil {
				return err
			}

			// Must provide non-zero index to prevent blocking
			// Index 1 is impossible anyways (due to Raft internals)
			if index == 0 {
				reply.Index = 1
			} else {
				reply.Index = index
			}

			if acl != nil {
				keys = FilterKeys(acl, keys)
			}
			reply.Keys = keys
			return nil
		})
}