/*
Copyright 2011 Google 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 memcache provides a client for the memcached cache server.
package memcache

import (
	"bufio"
	"bytes"
	"errors"
	"fmt"
	"io"
	"io/ioutil"
	"log"
	"net"
	"strconv"
	"strings"
	"sync"
	"time"

	"context"
)

// Similar to:
// http://code.google.com/appengine/docs/go/memcache/reference.html

var (
	// ErrCacheMiss means that a Get failed because the item wasn't present.
	ErrCacheMiss = &resumableError{
		message:     "memcache: cache miss",
		isResumable: true, // This response will not close the client connection.
	}

	// ErrCASConflict means that a CompareAndSwap call failed due to the
	// cached value being modified between the Get and the CompareAndSwap.
	// If the cached value was simply evicted rather than replaced,
	// ErrNotStored will be returned instead.
	ErrCASConflict = &resumableError{
		message:     "memcache: compare-and-swap conflict",
		isResumable: true, // This response will not close the client connection.
	}

	// ErrNotStored means that a conditional write operation (i.e. Add or
	// CompareAndSwap) failed because the condition was not satisfied.
	ErrNotStored = &resumableError{
		message:     "memcache: item not stored",
		isResumable: true, // This response will not close the client connection.
	}

	// ErrServer means that a server error occurred.
	ErrServerError = errors.New("memcache: server error")

	// ErrNoStats means that no statistics were available.
	ErrNoStats = errors.New("memcache: no statistics available")

	// ErrMalformedKey is returned when an invalid key is used.
	// Keys must be at maximum 250 bytes long, ASCII, and not
	// contain whitespace or control characters.
	ErrMalformedKey = &resumableError{
		message:     "malformed: key is too long or contains invalid characters",
		isResumable: true, // This response will not close the client connection.
	}

	// ErrNoServers is returned when no servers are configured or available.
	ErrNoServers = errors.New("memcache: no servers configured or available")
)

// objectTooLargeError is returned when the client tried to set an object in the
// cache, but the object was too large given the server max size setting.
func objectTooLargeError(bytes int) error {
	return &resumableError{
		message: "memcache: server error: object too large for cache (" +
			strconv.Itoa(bytes) + " bytes)",
		isResumable: true, // This response will not close the client connection.
	}
}

type resumable interface {
	IsResumable() bool
}

type resumableError struct {
	message     string
	isResumable bool
}

func (e *resumableError) Error() string {
	if e == nil {
		return "<nil>"
	}

	return e.message
}

func (e *resumableError) IsResumable() bool {
	if e == nil {
		return false
	}

	return e.isResumable
}

var _ resumable = (*resumableError)(nil)
var _ error = (*resumableError)(nil)

const (
	// DefaultTimeout is the default socket read/write timeout.
	DefaultTimeout = 100 * time.Millisecond

	// DefaultMaxIdleConns is the default maximum number of idle connections
	// kept for any single address.
	DefaultMaxIdleConns = 2

	buffered = 8 // arbitrary buffered channel size, for readability
)

// resumableError returns true if err is only a protocol-level cache error.
// This is used to determine whether or not a server connection should
// be re-used or not. If an error occurs, by default we don't reuse the
// connection, unless it was just a cache error.
func isResumable(err error) bool {
	if rErr, ok := err.(resumable); ok {
		return rErr.IsResumable()
	}

	return false
}

func legalKey(key string) bool {
	if len(key) > 250 {
		return false
	}
	for i := 0; i < len(key); i++ {
		if key[i] <= ' ' || key[i] > 0x7e {
			return false
		}
	}
	return true
}

var (
	crlf            = []byte("\r\n")
	space           = []byte(" ")
	resultOK        = []byte("OK\r\n")
	resultStored    = []byte("STORED\r\n")
	resultNotStored = []byte("NOT_STORED\r\n")
	resultExists    = []byte("EXISTS\r\n")
	resultNotFound  = []byte("NOT_FOUND\r\n")
	resultDeleted   = []byte("DELETED\r\n")
	resultEnd       = []byte("END\r\n")
	resultOk        = []byte("OK\r\n")
	resultTouched   = []byte("TOUCHED\r\n")

	resultClientErrorPrefix   = []byte("CLIENT_ERROR ")
	resultServerErrorTooLarge = []byte("SERVER_ERROR object too large for cache\r\n")
)

// New returns a memcache client using the provided server(s)
// with equal weight. If a server is listed multiple times,
// it gets a proportional amount of weight.
func New(server ...string) *Client {
	ss := &ServerList{}
	if err := ss.SetServers(server...); err != nil {
		return nil
	}
	return (&Client{selector: ss}).MaxIdleConns(0)
}

// Elasticache returns a memcache client which uses Elasticache's
// autodiscovery protocol to detect servers, and a stop function to
// cancel the autodiscovery polling.
func Elasticache(cfgServer string, pollInterval time.Duration) (*Client, func()) {
	ss := &ServerList{}
	c := &Client{selector: ss}

	var addr net.Addr
	addr, err := net.ResolveTCPAddr("tcp", cfgServer)
	if err != nil {
		panic(fmt.Errorf("Invalid TCP address: %s", cfgServer))
	}
	addr = newStaticAddr(addr)

	version := -1
	updateServers := func() {
		nc, err := c.dial(addr)
		if err != nil {
			log.Printf("Could not dial config server: %s", err)
			return
		}
		defer func() {
			if errr := nc.Close(); errr != nil {
				log.Printf("Could not close cache config server connection: %s", errr)
			}
		}()

		rw := bufio.NewReadWriter(bufio.NewReader(nc), bufio.NewWriter(nc))
		v, servers, err := getElasticacheServers(rw)
		if err != nil {
			log.Printf("Could not `config get cluster`: %s", err)
			return
		}

		if v > version {
			log.Printf("Cache servers (v%d): %v", v, servers)
			version = v
			if err := ss.SetServers(servers...); err != nil {
				log.Printf("Could not SetServers: %s", err)
			}
			c.initPool()
		}
	}

	updateServers()
	t := time.NewTicker(pollInterval)
	go func() {
		for range t.C {
			updateServers()
		}
	}()

	return c, t.Stop
}

// The output format that we're reading is described at
// http://docs.aws.amazon.com/AmazonElastiCache/latest/UserGuide/AutoDiscovery.AddingToYourClientLibrary.html#AutoDiscovery.AddingToYourClientLibrary.OutputFormat
func getElasticacheServers(rw *bufio.ReadWriter) (int, []string, error) {
	if _, err := fmt.Fprintf(rw, "config get cluster\r\n"); err != nil {
		return 0, nil, err
	}
	if err := rw.Flush(); err != nil {
		return 0, nil, err
	}
	var (
		key   string
		flags int
		size  int
	)
	if _, err := fmt.Fscanf(rw, "CONFIG %s %d %d\r\n", &key, &flags, &size); err != nil {
		return 0, nil, err
	}
	if key != "cluster" || flags != 0 {
		return 0, nil, fmt.Errorf("Unknown key or flags. key=%q flags=%d", key, flags)
	}
	value, err := ioutil.ReadAll(io.LimitReader(rw, int64(size)+2))
	if err != nil {
		return 0, nil, err
	}
	p := strings.SplitN(string(value), "\n", 3)
	if len(p) < 3 || p[2] != "\r\n" {
		return 0, nil, fmt.Errorf("Malformed response: %s", string(value))
	}
	version, err := strconv.Atoi(p[0])
	if err != nil {
		return 0, nil, err
	}
	var servers []string
	for _, s := range strings.Split(p[1], " ") {
		p := strings.Split(s, "|")
		servers = append(servers, p[1]+":"+p[2])
	}
	return version, servers, nil
}

// MaxIdleConns sets the per-address connection pool size limit, then re-initializes the pool
func (c *Client) MaxIdleConns(mic int) *Client {
	if mic <= 0 {
		mic = DefaultMaxIdleConns
	}

	c.mutex.Lock()
	c.maxIdleConns = mic
	c.mutex.Unlock()

	c.initPool()
	return c
}

// Prewarm sets the minimum number of idle connections, then re-initializes the pool
func (c *Client) Prewarm(s int) *Client {
	if s < 0 {
		s = 0
	}

	c.mutex.Lock()
	c.prewarmConns = s
	c.mutex.Unlock()

	c.initPool()
	return c
}

// initPool creates the freeconn map from the available servers.
// This map acts as a pool of idle connections for use in calls to the servers.
// initPool will always create a valid map with a max capacity of maxIdleConns.
// initPool will _try_ to copy existing idle connections into the new map.
// initPool will _try_ to create new idle connections up to the prewarm limit.
func (c *Client) initPool() {
	c.mutex.Lock()
	defer c.mutex.Unlock()

	oldconns := c.freeconn
	c.freeconn = map[string]chan *conn{}
	prewarm := c.prewarmConns
	if prewarm > c.maxIdleConns {
		prewarm = c.maxIdleConns
	}

	_ = c.selector.Each(func(a net.Addr) error {
		n := a.String()
		c.freeconn[n] = make(chan *conn, c.maxIdleConns)
		connsToWarm := prewarm

		if ch, ok := oldconns[n]; ok {
			close(ch)
			for conn := range ch {
				select {
				case c.freeconn[n] <- conn: // Copied successfully
					connsToWarm--
				default: // We likely decreased the MaxIdleConns so it doesn't fit any more
				}
			}
		}

		for i := 0; i < connsToWarm; i++ {
			nc, err := c.dial(a)
			if err != nil {
				continue
			}
			cn := &conn{
				nc:   nc,
				addr: a,
				rw:   bufio.NewReadWriter(bufio.NewReader(nc), bufio.NewWriter(nc)),
				c:    c,
			}

			select {
			case c.freeconn[n] <- cn: // Add the warm connection to the pool
			default: // We shouldn't ever hit this state, but better safe than sorry
			}
		}
		return nil
	})
}

// Close closes all idle connections and removes the idle pools
// This puts the client in a state where every request will create a new connection
// and any finished request will close the connection
// It does NOT close connections currently in use, but they will be closed when they complete
func (c *Client) Close() {
	c.mutex.Lock()
	defer c.mutex.Unlock()

	for _, ch := range c.freeconn {
	ClearChannel:
		for {
			select {
			case conn := <-ch:
				_ = conn.nc.Close()
			default:
				break ClearChannel
			}
		}
		close(ch)
	}
	c.freeconn = map[string]chan *conn{}
}

// Client is a memcache client.
// It is safe for unlocked use by multiple concurrent goroutines.
type Client struct {
	// Timeout specifies the socket read/write timeout.
	// If zero, DefaultTimeout is used.
	Timeout time.Duration

	selector ServerSelector

	mutex        sync.RWMutex
	freeconn     map[string]chan *conn
	maxIdleConns int
	prewarmConns int
}

// Item is an item to be got or stored in a memcached server.
type Item struct {
	// Key is the Item's key (250 bytes maximum).
	Key string

	// Value is the Item's value.
	Value []byte

	// Object is the Item's value for use with a Codec.
	Object interface{}

	// Flags are server-opaque flags whose semantics are entirely
	// up to the app.
	Flags uint32

	// Expiration is the cache expiration time, in seconds: either a relative
	// time from now (up to 1 month), or an absolute Unix epoch time.
	// Zero means the Item has no expiration time.
	Expiration int32

	// Compare and swap ID.
	casid uint64
}

// conn is a connection to a server.
type conn struct {
	nc   net.Conn
	rw   *bufio.ReadWriter
	addr net.Addr
	c    *Client
}

// release returns this connection back to the client's free pool
func (cn *conn) release() {
	cn.c.putFreeConn(cn.addr, cn)
}

func (cn *conn) extendDeadline() {
	_ = cn.nc.SetDeadline(time.Now().Add(cn.c.netTimeout()))
}

// condRelease releases this connection if the error pointed to by err
// is nil (not an error) or is only a protocol level error (e.g. a
// cache miss).  The purpose is to not recycle TCP connections that
// are bad.
func (cn *conn) condRelease(err *error) {
	if *err == nil || isResumable(*err) {
		cn.release()
	} else {
		_ = cn.nc.Close()
	}
}

func (c *Client) putFreeConn(addr net.Addr, cn *conn) {
	c.mutex.RLock()
	defer c.mutex.RUnlock()

	select {
	case c.freeconn[addr.String()] <- cn:
		// The connection was successfully added back to the pool
	default:
		// The pool is full, close the connection
		_ = cn.nc.Close()
	}
}

func (c *Client) getFreeConn(addr net.Addr) (cn *conn, ok bool) {
	c.mutex.RLock()
	defer c.mutex.RUnlock()

	select {
	case cn = <-c.freeconn[addr.String()]:
		return cn, true
	default:
		return nil, false
	}
}

func (c *Client) netTimeout() time.Duration {
	if c.Timeout != 0 {
		return c.Timeout
	}
	return DefaultTimeout
}

// ConnectTimeoutError is the error type used when it takes
// too long to connect to the desired host. This level of
// detail can generally be ignored.
type ConnectTimeoutError struct {
	Addr net.Addr
}

func (cte *ConnectTimeoutError) Error() string {
	return "memcache: connect timeout to " + cte.Addr.String()
}

func (c *Client) dial(addr net.Addr) (net.Conn, error) {
	nc, err := net.DialTimeout(addr.Network(), addr.String(), c.netTimeout())
	if err == nil {
		return nc, nil
	}

	if ne, ok := err.(net.Error); ok && ne.Timeout() {
		return nil, &ConnectTimeoutError{addr}
	}

	return nil, err
}

func (c *Client) getConn(addr net.Addr) (*conn, error) {
	cn, ok := c.getFreeConn(addr)
	if ok {
		cn.extendDeadline()
		return cn, nil
	}
	nc, err := c.dial(addr)
	if err != nil {
		return nil, err
	}
	cn = &conn{
		nc:   nc,
		addr: addr,
		rw:   bufio.NewReadWriter(bufio.NewReader(nc), bufio.NewWriter(nc)),
		c:    c,
	}
	cn.extendDeadline()
	return cn, nil
}

func (c *Client) onItem(item *Item, fn func(*Client, *bufio.ReadWriter, *Item) error) error {
	addr, err := c.selector.PickServer(item.Key)
	if err != nil {
		return err
	}
	cn, err := c.getConn(addr)
	if err != nil {
		return err
	}
	defer cn.condRelease(&err)
	err = fn(c, cn.rw, item) // Note we are setting the same err here, as pointed to above
	return err
}

// onItemAll is best effort. If the serverList gets modified when this function is running, then
// new servers data might not get written to new servers
func (c *Client) onItemAll(item *Item, fn func(*Client, *bufio.ReadWriter, *Item) error) []error {
	addrs := c.selector.GetServers()
	errs := []error{}
	channel := make(chan error, len(addrs))

	for _, addr := range addrs {
		go func(addr net.Addr) {
			defer func() {
				if p := recover(); p != nil {
					channel <- fmt.Errorf("gomemcache panic=%v", p)
				}
			}()
			cn, err := c.getConn(addr)
			if err != nil {
				channel <- err
			}
			defer cn.condRelease(&err)
			channel <- fn(c, cn.rw, item)
		}(addr)
	}

	for i := 0; i < len(addrs); i++ {
		currErr := <-channel
		if currErr != nil {
			errs = append(errs, currErr)
		}
	}
	return errs
}

func (c *Client) FlushAll() error {
	return c.selector.Each(c.flushAllFromAddr)
}

// Get gets the item for the given key. ErrCacheMiss is returned for a
// memcache cache miss. The key must be at most 250 bytes in length.
func (c *Client) Get(ctx context.Context, key string) (item *Item, err error) {
	err = c.withKeyAddr(key, func(addr net.Addr) error {
		return c.getFromAddr(addr, []string{key}, func(it *Item) { item = it })
	})
	if err == nil && item == nil {
		err = ErrCacheMiss
	}
	return
}

// GetFromRandom attempts to get the item for given key from a randomly
// chosen memcache server. ErrCacheMiss is returned for a
// memcache cache miss. The key must be at most 250 bytes in length.
func (c *Client) GetFromRandom(ctx context.Context, key string) (item *Item, err error) {
	err = c.withRandAddr(key, func(addr net.Addr) error {
		return c.getFromAddr(addr, []string{key}, func(it *Item) { item = it })
	})
	if err == nil && item == nil {
		err = ErrCacheMiss
	}
	return
}

// Touch updates the expiry for the given key. The seconds parameter is either
// a Unix timestamp or, if seconds is less than 1 month, the number of seconds
// into the future at which time the item will expire. ErrCacheMiss is returned if the
// key is not in the cache. The key must be at most 250 bytes in length.
func (c *Client) Touch(ctx context.Context, key string, seconds int32) (err error) {
	return c.withKeyAddr(key, func(addr net.Addr) error {
		return c.touchFromAddr(addr, []string{key}, seconds)
	})
}

func (c *Client) withKeyAddr(key string, fn func(net.Addr) error) (err error) {
	if !legalKey(key) {
		return ErrMalformedKey
	}
	addr, err := c.selector.PickServer(key)
	if err != nil {
		return err
	}
	return fn(addr)
}

func (c *Client) withRandAddr(key string, fn func(net.Addr) error) (err error) {
	if !legalKey(key) {
		return ErrMalformedKey
	}
	addr, err := c.selector.PickRandomServer(key)
	if err != nil {
		return err
	}
	return fn(addr)
}

func (c *Client) withAddrRw(addr net.Addr, fn func(*bufio.ReadWriter) error) (err error) {
	cn, err := c.getConn(addr)
	if err != nil {
		return err
	}
	defer cn.condRelease(&err)
	return fn(cn.rw)
}

func (c *Client) withKeyRw(key string, fn func(*bufio.ReadWriter) error) error {
	return c.withKeyAddr(key, func(addr net.Addr) error {
		return c.withAddrRw(addr, fn)
	})
}

func (c *Client) getFromAddr(addr net.Addr, keys []string, cb func(*Item)) error {
	return c.withAddrRw(addr, func(rw *bufio.ReadWriter) error {
		if _, err := fmt.Fprintf(rw, "gets %s\r\n", strings.Join(keys, " ")); err != nil {
			return err
		}
		if err := rw.Flush(); err != nil {
			return err
		}
		return parseGetResponse(rw.Reader, cb)
	})
}

// flushAllFromAddr send the flush_all command to the given addr
func (c *Client) flushAllFromAddr(addr net.Addr) error {
	return c.withAddrRw(addr, func(rw *bufio.ReadWriter) error {
		if _, err := fmt.Fprintf(rw, "flush_all\r\n"); err != nil {
			return err
		}
		if err := rw.Flush(); err != nil {
			return err
		}
		line, err := rw.ReadSlice('\n')
		if err != nil {
			return err
		}
		switch {
		case bytes.Equal(line, resultOk):
			break
		default:
			return fmt.Errorf("memcache: unexpected response line from flush_all: %q", string(line))
		}
		return nil
	})
}

func (c *Client) touchFromAddr(addr net.Addr, keys []string, expiration int32) error {
	return c.withAddrRw(addr, func(rw *bufio.ReadWriter) error {
		for _, key := range keys {
			if _, err := fmt.Fprintf(rw, "touch %s %d\r\n", key, expiration); err != nil {
				return err
			}
			if err := rw.Flush(); err != nil {
				return err
			}
			line, err := rw.ReadSlice('\n')
			if err != nil {
				return err
			}
			switch {
			case bytes.Equal(line, resultTouched):
				break // TODO: ineffective break statement. What are we supposed to do here?
			case bytes.Equal(line, resultNotFound):
				return ErrCacheMiss
			default:
				return fmt.Errorf("memcache: unexpected response line from touch: %q", string(line))
			}
		}
		return nil
	})
}

// GetMulti is a batch version of Get. The returned map from keys to
// items may have fewer elements than the input slice, due to memcache
// cache misses. Each key must be at most 250 bytes in length.
// If no error is returned, the returned map will also be non-nil.
func (c *Client) GetMulti(ctx context.Context, keys []string) (map[string]*Item, error) {
	var lk sync.Mutex
	m := make(map[string]*Item)
	addItemToMap := func(it *Item) {
		lk.Lock()
		defer lk.Unlock()
		m[it.Key] = it
	}

	keyMap := make(map[net.Addr][]string)
	for _, key := range keys {
		if !legalKey(key) {
			return nil, ErrMalformedKey
		}
		addr, err := c.selector.PickServer(key)
		if err != nil {
			return nil, err
		}
		keyMap[addr] = append(keyMap[addr], key)
	}

	ch := make(chan error, buffered)
	for addr, keys := range keyMap {
		go func(addr net.Addr, keys []string) {
			ch <- c.getFromAddr(addr, keys, addItemToMap)
		}(addr, keys)
	}

	var err error
	for range keyMap {
		if ge := <-ch; ge != nil {
			err = ge
		}
	}
	return m, err
}

// parseGetResponse reads a GET response from r and calls cb for each
// read and allocated Item
func parseGetResponse(r *bufio.Reader, cb func(*Item)) error {
	for {
		line, err := r.ReadSlice('\n')
		if err != nil {
			return err
		}
		if bytes.Equal(line, resultEnd) {
			return nil
		}
		it := new(Item)
		size, err := scanGetResponseLine(line, it)
		if err != nil {
			return err
		}
		it.Value, err = ioutil.ReadAll(io.LimitReader(r, int64(size)+2))
		if err != nil {
			return err
		}
		if !bytes.HasSuffix(it.Value, crlf) {
			return fmt.Errorf("memcache: corrupt get result read")
		}
		it.Value = it.Value[:size]
		cb(it)
	}
}

// scanGetResponseLine populates it and returns the declared size of the item.
// It does not read the bytes of the item.
func scanGetResponseLine(line []byte, it *Item) (size int, err error) {
	pattern := "VALUE %s %d %d %d\r\n"
	dest := []interface{}{&it.Key, &it.Flags, &size, &it.casid}
	if bytes.Count(line, space) == 3 {
		pattern = "VALUE %s %d %d\r\n"
		dest = dest[:3]
	}
	n, err := fmt.Sscanf(string(line), pattern, dest...)
	if err != nil || n != len(dest) {
		return -1, fmt.Errorf("memcache: unexpected line in get response: %q", line)
	}
	return size, nil
}

// Set writes the given item, unconditionally.
func (c *Client) Set(ctx context.Context, item *Item) error {
	return c.onItem(item, (*Client).set)
}

func (c *Client) set(rw *bufio.ReadWriter, item *Item) error {
	return c.populateOne(rw, "set", item)
}

// Best effort of setting the given item to all the memcache nodes, unconditionally.
// Returns a list of errors if some or all the writes fail.
func (c *Client) SetAll(ctx context.Context, item *Item) []error {
	return c.onItemAll(item, (*Client).set)
}

// Add writes the given item, if no value already exists for its
// key. ErrNotStored is returned if that condition is not met.
func (c *Client) Add(ctx context.Context, item *Item) error {
	return c.onItem(item, (*Client).add)
}

func (c *Client) add(rw *bufio.ReadWriter, item *Item) error {
	return c.populateOne(rw, "add", item)
}

// Best effort of adding the given item to all the memcache nodes, if no value already exists for its key.
// Returns a list of errors if some or all the writes fail (including ErrNotStored).
func (c *Client) AddAll(ctx context.Context, item *Item) []error {
	return c.onItemAll(item, (*Client).add)
}

// Replace writes the given item, but only if the server *does*
// already hold data for this key
func (c *Client) Replace(ctx context.Context, item *Item) error {
	return c.onItem(item, (*Client).replace)
}

func (c *Client) replace(rw *bufio.ReadWriter, item *Item) error {
	return c.populateOne(rw, "replace", item)
}

// Best effort of replacing the given item to all the memcache nodes, if the valueno value already exists for its key.
// Returns a list of errors if some or all the writes fail (including ErrNotStored).
func (c *Client) ReplaceAll(ctx context.Context, item *Item) []error {
	return c.onItemAll(item, (*Client).replace)
}

// CompareAndSwap writes the given item that was previously returned
// by Get, if the value was neither modified or evicted between the
// Get and the CompareAndSwap calls. The item's Key should not change
// between calls but all other item fields may differ. ErrCASConflict
// is returned if the value was modified in between the
// calls. ErrNotStored is returned if the value was evicted in between
// the calls.
func (c *Client) CompareAndSwap(ctx context.Context, item *Item) error {
	return c.onItem(item, (*Client).cas)
}

func (c *Client) cas(rw *bufio.ReadWriter, item *Item) error {
	return c.populateOne(rw, "cas", item)
}

func (c *Client) populateOne(rw *bufio.ReadWriter, verb string, item *Item) error {
	if !legalKey(item.Key) {
		return ErrMalformedKey
	}

	_, err := fmt.Fprintf(rw, "%s %s %d %d %d",
		verb, item.Key, item.Flags, item.Expiration, len(item.Value))
	if err != nil {
		return err
	}

	if verb == "cas" {
		id := strconv.FormatUint(item.casid, 10)
		if _, err = fmt.Fprint(rw, " "+id); err != nil {
			return err
		}
	}

	if _, err = rw.Write(crlf); err != nil {
		return err
	}

	if _, err = rw.Write(item.Value); err != nil {
		return err
	}

	if _, err = rw.Write(crlf); err != nil {
		return err
	}

	if err = rw.Flush(); err != nil {
		return err
	}
	line, err := rw.ReadSlice('\n')
	if err != nil {
		return err
	}

	switch {
	case bytes.Equal(line, resultStored):
		return nil
	case bytes.Equal(line, resultNotStored):
		return ErrNotStored
	case bytes.Equal(line, resultExists):
		return ErrCASConflict
	case bytes.Equal(line, resultNotFound):
		return ErrCacheMiss
	case bytes.Equal(line, resultServerErrorTooLarge):
		return objectTooLargeError(len(item.Value))
	}

	return fmt.Errorf("memcache: unexpected response line from %q: %q", verb, string(line))
}

func writeReadLine(rw *bufio.ReadWriter, format string, args ...interface{}) ([]byte, error) {
	_, err := fmt.Fprintf(rw, format, args...)
	if err != nil {
		return nil, err
	}
	if err = rw.Flush(); err != nil {
		return nil, err
	}
	line, err := rw.ReadSlice('\n')
	return line, err
}

func writeExpectf(rw *bufio.ReadWriter, expect []byte, format string, args ...interface{}) error {
	line, err := writeReadLine(rw, format, args...)
	if err != nil {
		return err
	}
	switch {
	case bytes.Equal(line, resultOK):
		return nil
	case bytes.Equal(line, expect):
		return nil
	case bytes.Equal(line, resultNotStored):
		return ErrNotStored
	case bytes.Equal(line, resultExists):
		return ErrCASConflict
	case bytes.Equal(line, resultNotFound):
		return ErrCacheMiss
	}
	return fmt.Errorf("memcache: unexpected response line: %q", string(line))
}

// Delete deletes the item with the provided key. The error ErrCacheMiss is
// returned if the item didn't already exist in the cache.
func (c *Client) Delete(ctx context.Context, key string) error {
	return c.withKeyRw(key, func(rw *bufio.ReadWriter) error {
		return writeExpectf(rw, resultDeleted, "delete %s\r\n", key)
	})
}

// DeleteAll deletes all items in the cache.
func (c *Client) DeleteAll(ctx context.Context) error {
	return c.withKeyRw("", func(rw *bufio.ReadWriter) error {
		return writeExpectf(rw, resultDeleted, "flush_all\r\n")
	})
}

// Increment atomically increments key by delta. The return value is
// the new value after being incremented or an error. If the value
// didn't exist in memcached the error is ErrCacheMiss. The value in
// memcached must be an decimal number, or an error will be returned.
// On 64-bit overflow, the new value wraps around.
func (c *Client) Increment(ctx context.Context, key string, delta uint64) (newValue uint64, err error) {
	return c.incrDecr("incr", key, delta)
}

// Decrement atomically decrements key by delta. The return value is
// the new value after being decremented or an error. If the value
// didn't exist in memcached the error is ErrCacheMiss. The value in
// memcached must be an decimal number, or an error will be returned.
// On underflow, the new value is capped at zero and does not wrap
// around.
func (c *Client) Decrement(ctx context.Context, key string, delta uint64) (newValue uint64, err error) {
	return c.incrDecr("decr", key, delta)
}

func (c *Client) incrDecr(verb, key string, delta uint64) (uint64, error) {
	var val uint64
	err := c.withKeyRw(key, func(rw *bufio.ReadWriter) error {
		line, err := writeReadLine(rw, "%s %s %d\r\n", verb, key, delta)
		if err != nil {
			return err
		}
		switch {
		case bytes.Equal(line, resultNotFound):
			return ErrCacheMiss
		case bytes.HasPrefix(line, resultClientErrorPrefix):
			errMsg := line[len(resultClientErrorPrefix) : len(line)-2]
			return errors.New("memcache: client error: " + string(errMsg))
		}
		val, err = strconv.ParseUint(string(line[:len(line)-2]), 10, 64)
		return err
	})
	return val, err
}
