kv.go

// Copyright 2014 Orchestrate, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package gorc

import (
	"bytes"
	"encoding/json"
	"fmt"
	"io"
	"io/ioutil"
	"net/url"
	"strconv"
	"strings"
)

// Holds results returned from a KV list query.
type KVResults struct {
	Count   uint64     `json:"count"`
	Results []KVResult `json:"results"`
	Next    string     `json:"next,omitempty"`
}

// An individual Key/Value result.
type KVResult struct {
	Path     Path            `json:"path"`
	RawValue json.RawMessage `json:"value"`
}

// Represents a single operation to be performed when patching an existing
// object. Each operation can mutate the data in some way, or test that the
// data is in a specific state.
type PatchOperation struct {
	// The operation to perform. This is required and must be a known value
	// from the list of Ops defined in this package.
	Op string `json:"op"`

	Path string `json:"path"`

	// The value to use when performing the operation.
	Value interface{} `json:"value,omitempty"`
}

// Represents a set PatchOperations that should be applied to a specific
// item. These will all be performed together, and if any test case
// fails then no mutations will happen at all.
type PatchSet []PatchOperation

// Get a collection-key pair's value.
func (c *Client) Get(collection, key string) (*KVResult, error) {
	return c.GetPath(&Path{Collection: collection, Key: key})
}

// Get the value at a path.
func (c *Client) GetPath(path *Path) (*KVResult, error) {
	resp, err := c.doRequest("GET", path.trailingGetURI(), nil, nil)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()

	// If the request ended in error then read the body into an
	// OrchestrateError object.
	if resp.StatusCode != 200 {
		return nil, newError(resp)
	}

	// Decode the body into a JSON object.
	// TODO: Check for a content-length header so we can pre-allocate buffer
	// space.
	buf := bytes.NewBuffer(nil)
	if _, err := buf.ReadFrom(resp.Body); err != nil {
		return nil, err
	}

	if path.Ref == "" {
		parts := strings.SplitAfter(resp.Header.Get("Content-Location"), "/")
		if len(parts) >= 6 {
			path.Ref = parts[5]
		}
	}

	return &KVResult{Path: *path, RawValue: buf.Bytes()}, nil
}

// Store a value to a collection-key pair.
func (c *Client) Put(collection string, key string, value interface{}) (*Path, error) {
	reader, writer := io.Pipe()
	encoder := json.NewEncoder(writer)

	go func() { writer.CloseWithError(encoder.Encode(value)) }()
	return c.PutRaw(collection, key, reader)
}

// Store a value to a collection-key pair.
func (c *Client) PutRaw(collection string, key string, value io.Reader) (*Path, error) {
	return c.doPut(&Path{Collection: collection, Key: key}, nil, value)
}

// Store a value to a collection-key pair if the path's ref value is the latest.
func (c *Client) PutIfUnmodified(path *Path, value interface{}) (*Path, error) {
	reader, writer := io.Pipe()
	encoder := json.NewEncoder(writer)

	go func() { writer.CloseWithError(encoder.Encode(value)) }()
	return c.PutIfUnmodifiedRaw(path, reader)
}

// Store a value to a collection-key pair if the path's ref value is the latest.
func (c *Client) PutIfUnmodifiedRaw(path *Path, value io.Reader) (*Path, error) {
	headers := map[string]string{
		"If-Match": `"` + path.Ref + `"`,
	}

	return c.doPut(path, headers, value)
}

// Store a value to a collection-key pair if it doesn't already hold a value.
func (c *Client) PutIfAbsent(collection, key string, value interface{}) (*Path, error) {
	reader, writer := io.Pipe()
	encoder := json.NewEncoder(writer)

	go func() { writer.CloseWithError(encoder.Encode(value)) }()
	return c.PutIfAbsentRaw(collection, key, reader)
}

// Store a value to a collection-key pair if it doesn't already hold a value.
func (c *Client) PutIfAbsentRaw(collection, key string, value io.Reader) (*Path, error) {
	headers := map[string]string{
		"If-None-Match": "\"*\"",
	}

	return c.doPut(&Path{Collection: collection, Key: key}, headers, value)
}

// Execute a key/value Put.
func (c *Client) doPut(path *Path, headers map[string]string, value io.Reader) (*Path, error) {
	resp, err := c.doRequest("PUT", path.trailingPutURI(), headers, value)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()

	// If the request ended in error then read the body into an
	// OrchestrateError object.
	if resp.StatusCode != 201 {
		return nil, newError(resp)
	}

	// Read the body so the connection can be properly reused.
	io.Copy(ioutil.Discard, resp.Body)

	// Parse the ref of the returned object.
	ref := ""
	parts := strings.SplitAfter(resp.Header.Get("Location"), "/")
	if len(parts) >= 6 {
		ref = parts[5]
	} else {
		return nil, fmt.Errorf("Missing ref component: %s", resp.Header.Get("Location"))
	}

	// Return the results.
	return &Path{
		Collection: path.Collection,
		Key:        path.Key,
		Ref:        ref,
	}, err
}

// Send a set of patch operations for a collection-key pair.
func (c *Client) Patch(collection string, key string, value PatchSet) (*Path, error) {
	reader, writer := io.Pipe()
	encoder := json.NewEncoder(writer)

	go func() { writer.CloseWithError(encoder.Encode(value)) }()
	return c.PatchRaw(collection, key, reader)
}

// Send a set of patch operations for a collection-key pair.
func (c *Client) PatchRaw(collection string, key string, value io.Reader) (*Path, error) {
	header := map[string]string{
		"Content-Type": "application/json-patch+json",
	}
	return c.doPatch(&Path{Collection: collection, Key: key}, header, value)
}

// Execute a Patch with partial updates.
func (c *Client) doPatch(path *Path, headers map[string]string, value io.Reader) (*Path, error) {
	resp, err := c.doRequest("PATCH", path.trailingPutURI(), headers, value)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()

	// If the request ended in error then read the body into an
	// OrchestrateError object.
	if resp.StatusCode != 201 {
		return nil, newError(resp)
	}

	// Read the body so the connection can be properly reused.
	io.Copy(ioutil.Discard, resp.Body)

	// Parse the ref of the returned object.
	ref := ""
	parts := strings.SplitAfter(resp.Header.Get("Location"), "/")
	if len(parts) >= 6 {
		ref = parts[5]
	} else {
		return nil, fmt.Errorf("Missing ref component: %s", resp.Header.Get("Location"))
	}

	// Return the results.
	return &Path{
		Collection: path.Collection,
		Key:        path.Key,
		Ref:        ref,
	}, err
}

// Delete the value held at a collection-key pair.
func (c *Client) Delete(collection, key string) error {
	return c.doDelete(collection+"/"+key, nil)
}

// Delete the value held at a collection-key par if the path's ref value is the
// latest.
func (c *Client) DeleteIfUnmodified(path *Path) error {
	headers := map[string]string{
		"If-Match": `"` + path.Ref + `"`,
	}

	return c.doDelete(path.trailingPutURI(), headers)
}

// Delete the current and all previous values from a collection-key pair.
func (c *Client) Purge(collection, key string) error {
	return c.doDelete(collection+"/"+key+"?purge=true", nil)
}

// Delete a collection.
func (c *Client) DeleteCollection(collection string) error {
	return c.doDelete(collection+"?force=true", nil)
}

// Execute delete
func (c *Client) doDelete(trailingUri string, headers map[string]string) error {
	resp, err := c.doRequest("DELETE", trailingUri, headers, nil)
	if err != nil {
		return err
	}
	defer resp.Body.Close()

	// If the request ended in error then read the body into an
	// OrchestrateError object.
	if resp.StatusCode != 204 {
		return newError(resp)
	}

	// Read the body so the connection can be properly reused.
	io.Copy(ioutil.Discard, resp.Body)

	return nil
}

// List the values in a collection in key order with the specified page size.
func (c *Client) List(collection string, limit int) (*KVResults, error) {
	queryVariables := url.Values{
		"limit": []string{strconv.Itoa(limit)},
	}

	trailingUri := collection + "?" + queryVariables.Encode()

	return c.doList(trailingUri)
}

// List the values in a collection in key order with the specified page size
// that come after the specified key.
func (c *Client) ListAfter(collection, after string, limit int) (*KVResults, error) {
	queryVariables := url.Values{
		"limit":    []string{strconv.Itoa(limit)},
		"afterKey": []string{after},
	}

	trailingUri := collection + "?" + queryVariables.Encode()

	return c.doList(trailingUri)
}

// List the values in a collection in key order with the specified page size
// starting with the specified key.
func (c *Client) ListStart(collection, start string, limit int) (*KVResults, error) {
	queryVariables := url.Values{
		"limit":    []string{strconv.Itoa(limit)},
		"startKey": []string{start},
	}

	trailingUri := collection + "?" + queryVariables.Encode()

	return c.doList(trailingUri)
}

// List the values in a collection within a given range of keys, starting with the
// specified key and stopping at the end key
func (c *Client) ListRange(collection, start, end string, limit int) (*KVResults, error) {
	queryVariables := url.Values{
		"limit":    []string{strconv.Itoa(limit)},
		"startKey": []string{start},
		"endKey":   []string{end},
	}

	trailingUri := collection + "?" + queryVariables.Encode()

	return c.doList(trailingUri)
}

// Get the page of key/value list results that follow that provided set.
func (c *Client) ListGetNext(results *KVResults) (*KVResults, error) {
	return c.doList(results.Next[4:])
}

// Execute a key/value list operation.
func (c *Client) doList(trailingUri string) (*KVResults, error) {
	resp, err := c.doRequest("GET", trailingUri, nil, nil)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()

	// If the request ended in error then read the body into an
	// OrchestrateError object.
	if resp.StatusCode != 200 {
		return nil, newError(resp)
	}

	// Decode the returned JSON into the results.
	decoder := json.NewDecoder(resp.Body)
	result := new(KVResults)
	if err := decoder.Decode(result); err != nil {
		return result, err
	}

	return result, nil
}

// Check if there is a subsequent page of key/value list results.
func (r *KVResults) HasNext() bool {
	return r.Next != ""
}

// Marshall the value of a KVResult into the provided object.
func (r *KVResult) Value(value interface{}) error {
	return json.Unmarshal(r.RawValue, value)
}

// Returns the trailing URI part for a GET request.
func (p *Path) trailingGetURI() string {
	if p.Ref != "" {
		return p.Collection + "/" + p.Key + "/refs/" + p.Ref
	}
	return p.Collection + "/" + p.Key
}

// Returns the trailing URI part for a PUT request.
func (p *Path) trailingPutURI() string {
	return p.Collection + "/" + p.Key
}

// Replace appends a "Replace" Operation to the UdateSet. Replace operations work like
// add except that the value must exist prior to the call for the operation
// to succeed.
func (ps *PatchSet) Replace(path string, value interface{}) {
	*ps = append(*ps, PatchOperation{Op: "replace", Path: path, Value: value})
}

// Inc appends an "Inc" Operation to the UdateSet. Inc operations will increment a
// counter vie the given value. To decrement the value provide a negative
// value.
func (ps *PatchSet) Inc(path string, value float64) {
	*ps = append(*ps, PatchOperation{Op: "inc", Path: path, Value: value})
}

// Reset removes the patch operations added previously
func (ps *PatchSet) Reset() {
	*ps = make(PatchSet, 0)
}