verbs.go

// Copyright CattleCloud LLC 2025, 2026
// SPDX-License-Identifier: BSD-3-Clause

package memc

import (
	"bufio"
	"errors"
	"fmt"
	"io"
	"strconv"
	"strings"
	"time"

	"cattlecloud.net/go/memc/iopool"
)

var (
	ErrCacheMiss    = errors.New("memc: cache miss")
	ErrKeyNotValid  = errors.New("memc: key is not valid")
	ErrNotStored    = errors.New("memc: item not stored")
	ErrNotFound     = errors.New("memc: item not found")
	ErrConflict     = errors.New("memc: CAS conflict")
	ErrExpiration   = errors.New("memc: expiration ttl is not valid")
	ErrClientClosed = errors.New("memc: client has been closed")
	ErrNegativeInc  = errors.New("memc: increment delta must be non-negative")
	ErrNonNumeric   = errors.New("memc: cannot increment non-numeric value")
	ErrCommandIssue = errors.New("memc: got command error response")
)

// CAS represents a Compare-And-Swap token used for optimistic locking.
// It is returned by Gets and must be provided to CompareAndSwap to atomically update a value.
type CAS uint64

// Options contains configuration parameters that may be applied when executing
// a verb like Get, Set, etc.
type Options struct {
	expiration time.Duration
	flags      int
}

// Option to apply when executing a verb like Get, Set, etc.
type Option func(o *Options)

// TTL applies the given expiration time to set on the value being set.
//
// The expiration must be greater than 1 second, or 0, indicating the value will
// not expire automatically.
func TTL(expiration time.Duration) Option {
	return func(o *Options) {
		o.expiration = expiration
	}
}

// Flags applies the given flags on the value being set.
func Flags(flags int) Option {
	return func(o *Options) {
		o.flags = flags
	}
}

// Set will store the item using the given key, possibly overwriting any
// existing data. New items are at the top of the LRU.
//
// Uses Client c to connect to a memcached instance, and automatically handles
// connection pooling and reuse.
//
// One or more Option(s) may be applied to configure things such as the
// value expiration TTL or its associated flags.
func Set[T any](c *Client, key string, item T, opts ...Option) error {
	if err := check(key); err != nil {
		return err
	}

	options := &Options{
		expiration: c.expiration,
		flags:      0,
	}

	for _, opt := range opts {
		opt(options)
	}

	return c.do(key, func(conn *iopool.Buffer) error {
		encoding, encerr := encode(item)
		if encerr != nil {
			return encerr
		}

		expiration, experr := c.seconds(options.expiration)
		if experr != nil {
			return experr
		}

		// write the header components
		if _, err := fmt.Fprintf(
			conn,
			"set %s %d %d %d\r\n",
			key, options.flags, expiration, len(encoding),
		); err != nil {
			return err
		}

		// write the payload
		if _, err := conn.Write(encoding); err != nil {
			return err
		}

		// write clrf
		if _, err := io.WriteString(conn, "\r\n"); err != nil {
			return err
		}

		// flush the buffer
		if err := conn.Flush(); err != nil {
			return err
		}

		// read response
		line, lerr := conn.ReadSlice('\n')
		if lerr != nil {
			return lerr
		}

		switch string(line) {
		case "STORED\r\n":
			return nil
		case "NOT_STORED\r\n":
			return ErrNotStored
		default:
			return fmt.Errorf("memc: unexpected response to set: %q", string(line))
		}
	})
}

// Replace will store the item using the given key, but only if the key
// already exists. New items are at the top of the LRU.
//
// Uses Client c to connect to a memcached instance, and automatically handles
// connection pooling and reuse.
//
// One or more Option(s) may be applied to configure things such as the
// value expiration TTL or its associated flags.
func Replace[T any](c *Client, key string, item T, opts ...Option) error {
	if err := check(key); err != nil {
		return err
	}

	options := &Options{
		expiration: c.expiration,
		flags:      0,
	}

	for _, opt := range opts {
		opt(options)
	}

	return c.do(key, func(conn *iopool.Buffer) error {
		encoding, encerr := encode(item)
		if encerr != nil {
			return encerr
		}

		expiration, experr := c.seconds(options.expiration)
		if experr != nil {
			return experr
		}

		// write the header components
		if _, err := fmt.Fprintf(
			conn,
			"replace %s %d %d %d\r\n",
			key, options.flags, expiration, len(encoding),
		); err != nil {
			return err
		}

		// write the payload
		if _, err := conn.Write(encoding); err != nil {
			return err
		}

		// write clrf
		if _, err := io.WriteString(conn, "\r\n"); err != nil {
			return err
		}

		// flush the buffer
		if err := conn.Flush(); err != nil {
			return err
		}

		// read response
		line, lerr := conn.ReadSlice('\n')
		if lerr != nil {
			return lerr
		}

		switch string(line) {
		case "STORED\r\n":
			return nil
		case "NOT_STORED\r\n":
			return ErrNotStored
		default:
			return fmt.Errorf("memc: unexpected response to replace: %q", string(line))
		}
	})
}

// Prepend will prepend the given value to the value associated with the given key.
//
// Prepend differs from Set in that it is meant to add additional data to an
// existing key, rather than replace the existing value entirely. The key
// must already exist.
//
// Uses Client c to connect to a memcached instance, and automatically handles
// connection pooling and reuse.
//
// One or more Option(s) may be applied to configure things such as the
// value expiration TTL or its associated flags.
func Prepend[T any](c *Client, key string, item T, opts ...Option) error {
	if err := check(key); err != nil {
		return err
	}

	options := &Options{
		expiration: c.expiration,
		flags:      0,
	}

	for _, opt := range opts {
		opt(options)
	}

	return c.do(key, func(conn *iopool.Buffer) error {
		encoding, encerr := encode(item)
		if encerr != nil {
			return encerr
		}

		expiration, experr := c.seconds(options.expiration)
		if experr != nil {
			return experr
		}

		// write the header components
		if _, err := fmt.Fprintf(
			conn,
			"prepend %s %d %d %d\r\n",
			key, options.flags, expiration, len(encoding),
		); err != nil {
			return err
		}

		// write the payload
		if _, err := conn.Write(encoding); err != nil {
			return err
		}

		// write clrf
		if _, err := io.WriteString(conn, "\r\n"); err != nil {
			return err
		}

		// flush the buffer
		if err := conn.Flush(); err != nil {
			return err
		}

		// read response
		line, lerr := conn.ReadSlice('\n')
		if lerr != nil {
			return lerr
		}

		switch string(line) {
		case "STORED\r\n":
			return nil
		case "NOT_STORED\r\n":
			return ErrNotStored
		case "NOT_FOUND\r\n":
			return ErrNotFound
		default:
			return fmt.Errorf("memc: unexpected response to prepend: %q", string(line))
		}
	})
}

// Append will append the given value to the value associated with the given key.
//
// Append differs from Set in that it is meant to add additional data to an
// existing key, rather than replace the existing value entirely. The key
// must already exist.
//
// Uses Client c to connect to a memcached instance, and automatically handles
// connection pooling and reuse.
//
// One or more Option(s) may be applied to configure things such as the
// value expiration TTL or its associated flags.
func Append[T any](c *Client, key string, item T, opts ...Option) error {
	if err := check(key); err != nil {
		return err
	}

	options := &Options{
		expiration: c.expiration,
		flags:      0,
	}

	for _, opt := range opts {
		opt(options)
	}

	return c.do(key, func(conn *iopool.Buffer) error {
		encoding, encerr := encode(item)
		if encerr != nil {
			return encerr
		}

		expiration, experr := c.seconds(options.expiration)
		if experr != nil {
			return experr
		}

		// write the header components
		if _, err := fmt.Fprintf(
			conn,
			"append %s %d %d %d\r\n",
			key, options.flags, expiration, len(encoding),
		); err != nil {
			return err
		}

		// write the payload
		if _, err := conn.Write(encoding); err != nil {
			return err
		}

		// write clrf
		if _, err := io.WriteString(conn, "\r\n"); err != nil {
			return err
		}

		// flush the buffer
		if err := conn.Flush(); err != nil {
			return err
		}

		// read response
		line, lerr := conn.ReadSlice('\n')
		if lerr != nil {
			return lerr
		}

		switch string(line) {
		case "STORED\r\n":
			return nil
		case "NOT_STORED\r\n":
			return ErrNotStored
		case "NOT_FOUND\r\n":
			return ErrNotFound
		default:
			return fmt.Errorf("memc: unexpected response to append: %q", string(line))
		}
	})
}

// Add will store the item using the given key, but only if no item currently
// exists. New items are at the top of the LRU.
//
// Uses Client c to connect to a memcached instance, and automatically handles
// connection pooling and reuse.
//
// One or more Option(s) may be applied to configure things such as the
// value expiration TTL or its associated flags.
func Add[T any](c *Client, key string, item T, opts ...Option) error {
	if err := check(key); err != nil {
		return err
	}

	options := &Options{
		expiration: c.expiration,
		flags:      0,
	}

	for _, opt := range opts {
		opt(options)
	}

	return c.do(key, func(conn *iopool.Buffer) error {
		encoding, encerr := encode(item)
		if encerr != nil {
			return encerr
		}

		expiration, experr := c.seconds(options.expiration)
		if experr != nil {
			return experr
		}

		// write the header components
		if _, err := fmt.Fprintf(
			conn,
			"add %s %d %d %d\r\n",
			key, options.flags, expiration, len(encoding),
		); err != nil {
			return err
		}

		// write the payload
		if _, err := conn.Write(encoding); err != nil {
			return err
		}

		// write clrf
		if _, err := io.WriteString(conn, "\r\n"); err != nil {
			return err
		}

		// flush the buffer
		if err := conn.Flush(); err != nil {
			return err
		}

		// read response
		line, lerr := conn.ReadSlice('\n')
		if lerr != nil {
			return lerr
		}

		switch string(line) {
		case "STORED\r\n":
			return nil
		case "NOT_STORED\r\n":
			return ErrNotStored
		case "EXISTS\r\n":
			return ErrConflict
		default:
			return fmt.Errorf("memc: unexpected response to set: %q", string(line))
		}
	})
}

// CompareAndSwap will store the item using the given key, but only if the CAS
// token matches the current value's CAS token. This provides atomic
// compare-and-swap functionality for optimistic locking.
//
// If the key does not exist, ErrNotFound is returned.
//
// If the CAS token does not match (meaning the value was modified since it was
// retrieved with Gets), ErrConflict is returned.
//
// Uses Client c to connect to a memcached instance, and automatically handles
// connection pooling and reuse.
//
// One or more Option(s) may be applied to configure things such as the value
// expiration TTL or its associated flags.
func CompareAndSwap[T any](c *Client, key string, cas CAS, item T, opts ...Option) error {
	if err := check(key); err != nil {
		return err
	}

	options := &Options{
		expiration: c.expiration,
		flags:      0,
	}

	for _, opt := range opts {
		opt(options)
	}

	return c.do(key, func(conn *iopool.Buffer) error {
		encoding, encerr := encode(item)
		if encerr != nil {
			return encerr
		}

		expiration, experr := c.seconds(options.expiration)
		if experr != nil {
			return experr
		}

		// write the header components with CAS token
		if _, err := fmt.Fprintf(
			conn,
			"cas %s %d %d %d %d\r\n",
			key, options.flags, expiration, len(encoding), cas,
		); err != nil {
			return err
		}

		// write the payload
		if _, err := conn.Write(encoding); err != nil {
			return err
		}

		// write clrf
		if _, err := io.WriteString(conn, "\r\n"); err != nil {
			return err
		}

		// flush the buffer
		if err := conn.Flush(); err != nil {
			return err
		}

		// read response
		line, lerr := conn.ReadSlice('\n')
		if lerr != nil {
			return lerr
		}

		switch string(line) {
		case "STORED\r\n":
			return nil
		case "NOT_FOUND\r\n":
			return ErrNotFound
		case "EXISTS\r\n":
			return ErrConflict
		default:
			return fmt.Errorf("memc: unexpected response to cas: %q", string(line))
		}
	})
}

// Get the value associated with the given key.
//
// Uses Client c to connect to a memcached instance, and automatically handles
// connection pooling and reuse.
func Get[T any](c *Client, key string) (T, error) {
	var result T

	if err := check(key); err != nil {
		return result, err
	}

	err := c.do(key, func(conn *iopool.Buffer) error {
		// write the header components
		if _, err := fmt.Fprintf(conn, "get %s\r\n", key); err != nil {
			return err
		}

		// flush the connection, forcing bytes over the wire
		if err := conn.Flush(); err != nil {
			return err
		}

		// read the response payload
		payload, err := getPayload(conn.Reader)
		if err != nil {
			return err
		}

		result, err = decode[T](payload)
		return err
	})

	return result, err
}

// Gets the value associated with the given key, along with its CAS token.
//
// The CAS token can be used with CompareAndSwap to atomically update the value,
// providing optimistic locking. If the value has been modified since it was
// retrieved, CompareAndSwap will return an ErrConflict error.
//
// Uses Client c to connect to a memcached instance, and automatically handles
// connection pooling and reuse.
func Gets[T any](c *Client, key string) (T, CAS, error) {
	var result T
	var casToken CAS

	if err := check(key); err != nil {
		return result, 0, err
	}

	err := c.do(key, func(conn *iopool.Buffer) error {
		// write the header components
		if _, err := fmt.Fprintf(conn, "gets %s\r\n", key); err != nil {
			return err
		}

		// flush the connection, forcing bytes over the wire
		if err := conn.Flush(); err != nil {
			return err
		}

		// read the response payload with CAS token
		payload, cas, err := getPayloadWithCAS(conn.Reader)
		if err != nil {
			return err
		}

		result, err = decode[T](payload)
		if err != nil {
			return err
		}

		casToken = CAS(cas)
		return nil
	})

	return result, casToken, err
}

func getPayload(r *bufio.Reader) ([]byte, error) {
	b, err := r.ReadSlice('\n')
	if err != nil {
		return nil, err
	}

	// key was not found, is a cache miss
	if string(b) == "END\r\n" {
		return nil, ErrCacheMiss
	}

	expect := "VALUE %s %d %d\r\n"
	var (
		key   string
		flags int
		size  int
	)

	// scan the header line, giving us a payload size
	if _, err = fmt.Sscanf(string(b), expect, &key, &flags, &size); err != nil {
		return nil, err
	}

	// read the data into our payload
	payload := make([]byte, size+2) // including \r\n
	if _, err = io.ReadFull(r, payload); err != nil {
		return nil, err
	}
	payload = payload[0:size] // chop \r\n

	// read the trailing line ("END\r\n")
	b, err = r.ReadSlice('\n')
	if err != nil {
		return nil, err
	}
	if string(b) != "END\r\n" {
		return nil, unexpected(b)
	}

	return payload, err
}

func getPayloadWithCAS(r *bufio.Reader) ([]byte, uint64, error) {
	b, err := r.ReadSlice('\n')
	if err != nil {
		return nil, 0, err
	}

	// key was not found, is a cache miss
	if string(b) == "END\r\n" {
		return nil, 0, ErrCacheMiss
	}

	// handle CAS value - format is "VALUE key flags bytes cas\r\n"
	expect := "VALUE %s %d %d %d\r\n"
	var (
		key   string
		flags int
		size  int
		cas   uint64
	)

	// scan the header line, giving us a payload size and CAS token
	if _, err = fmt.Sscanf(string(b), expect, &key, &flags, &size, &cas); err != nil {
		return nil, 0, err
	}

	// read the data into our payload
	payload := make([]byte, size+2) // including \r\n
	if _, err = io.ReadFull(r, payload); err != nil {
		return nil, 0, err
	}
	payload = payload[0:size] // chop \r\n

	// read the trailing line ("END\r\n")
	b, err = r.ReadSlice('\n')
	if err != nil {
		return nil, 0, err
	}
	if string(b) != "END\r\n" {
		return nil, 0, unexpected(b)
	}

	return payload, cas, nil
}

// Flush will delete all items from memcached.
//
// The timeout parameter is optional. A timeout of 0 means flush right now.
// A non-zero timeout will delay the flush by that many seconds.
//
// Note: this operation is performed on a single memcached server, even when
// the Client is configured with multiple server addresses. This is intentional,
// as flush is typically used by local administration tools that connect to a
// single memcached instance.
func Flush(c *Client, timeout time.Duration) error {
	return c.do("", func(conn *iopool.Buffer) error {
		expiration, err := c.seconds(timeout)
		if err != nil {
			return err
		}

		if _, err := fmt.Fprintf(
			conn,
			"flush_all %d\r\n",
			expiration,
		); err != nil {
			return err
		}

		if err := conn.Flush(); err != nil {
			return err
		}

		line, lerr := conn.ReadSlice('\n')
		if lerr != nil {
			return lerr
		}

		switch string(line) {
		case "OK\r\n":
			return nil
		default:
			return unexpected(line)
		}
	})
}

// Delete will remove the value associated with key from memcached.
//
// Uses Client c to connect to a memcached instance, and automatically handles
// connection pooling and reuse.
func Delete(c *Client, key string) error {
	if err := check(key); err != nil {
		return err
	}

	return c.do(key, func(conn *iopool.Buffer) error {
		// write the header components
		if _, err := fmt.Fprintf(
			conn,
			"delete %s\r\n",
			key,
		); err != nil {
			return err
		}

		// flush the buffer
		if err := conn.Flush(); err != nil {
			return err
		}

		line, lerr := conn.ReadSlice('\n')
		if lerr != nil {
			return lerr
		}

		switch string(line) {
		case "DELETED\r\n":
			return nil
		case "NOT_FOUND\r\n":
			return ErrNotFound
		default:
			return unexpected(line)
		}
	})
}

// Increment will increment the value associated with the given key by delta.
//
// Note: the value must be an ASCII integer. It must have been initially stored
// as a string value, e.g. by using Set. The delta value must be positive.
//
//	Set(client, "counter", "100")
//	Increment(client, "counter", 1) // counter = 101
func Increment[T Countable](c *Client, key string, delta T) (T, error) {
	if err := check(key); err != nil {
		return T(0), err
	}

	if delta < 0 {
		return T(0), ErrNegativeInc
	}

	var result T

	err := c.do(key, func(conn *iopool.Buffer) error {
		// write the header components
		if _, err := fmt.Fprintf(
			conn,
			"incr %s %d\r\n",
			key, delta,
		); err != nil {
			return err
		}

		// flush the buffer
		if err := conn.Flush(); err != nil {
			return err
		}

		// read the response
		line, lerr := conn.ReadSlice('\n')
		if lerr != nil {
			return lerr
		}

		// check for error response
		s := string(line)
		switch {
		case s == "NOT_FOUND\r\n":
			return ErrNotFound
		case strings.Contains(s, "cannot increment or decrement non-numeric value"):
			return ErrNonNumeric
		}

		// parse response as the resulting value
		s = strings.TrimSpace(s)
		u, uerr := strconv.ParseUint(s, 10, 64)
		if uerr != nil {
			return unexpected(line)
		}

		// recast to value type
		result = T(u)

		return nil
	})

	return result, err
}

// Decrement will decrement the value associated with the given key by delta.
//
// Note: the value must be an ASCII integer. It must have been initially stored
// as a string value, e.g. by using Set. The delta value must be positive.
//
//	Set(client, "counter", "100")
//	Decrement(client, "counter", 1) // counter = 99
func Decrement[T Countable](c *Client, key string, delta T) (T, error) {
	if err := check(key); err != nil {
		return T(0), err
	}

	if delta < 0 {
		return T(0), ErrNegativeInc
	}

	var result T

	err := c.do(key, func(conn *iopool.Buffer) error {
		// write the header components
		if _, err := fmt.Fprintf(
			conn,
			"decr %s %d\r\n",
			key, delta,
		); err != nil {
			return err
		}

		// flush the buffer
		if err := conn.Flush(); err != nil {
			return err
		}

		// read the response
		line, lerr := conn.ReadSlice('\n')
		if lerr != nil {
			return lerr
		}

		// check for error response
		s := string(line)
		switch {
		case s == "NOT_FOUND\r\n":
			return ErrNotFound
		case strings.Contains(s, "cannot increment or decrement non-numeric value"):
			return ErrNonNumeric
		}

		// parse response as the resulting value
		s = strings.TrimSpace(s)
		u, uerr := strconv.ParseUint(s, 10, 64)
		if uerr != nil {
			return unexpected(line)
		}

		// recast to value type
		result = T(u)

		return nil
	})

	return result, err
}

// Stats returns runtime statistics for a single memcached server.
//
// Note: this operation is performed on a single memcached server, even when
// the Client is configured with multiple server addresses. This is intentional,
// as stats is typically used by local monitoring tools that connect to a
// single memcached instance.
func Stats(c *Client) (*Statistics, error) {
	var statistics *Statistics

	err := c.do("", func(conn *iopool.Buffer) error {
		// write the header component
		if _, err := fmt.Fprintf(conn, "stats\r\n"); err != nil {
			return err
		}

		// flush the connection, forcing bytes over the wire
		if err := conn.Flush(); err != nil {
			return err
		}

		// extract the statistics payload
		payload, perr := stats(conn.Reader)
		if perr != nil {
			return perr
		}
		statistics = payload

		return nil
	})

	return statistics, err
}

// StatsSlabs returns slab statistics for a single memcached server.
//
// Note: this operation is performed on a single memcached server, even when
// the Client is configured with multiple server addresses. This is intentional,
// as stats is typically used by local monitoring tools that connect to a
// single memcached instance.
func StatsSlabs(c *Client) (*SlabStatistics, error) {
	var statistics *SlabStatistics

	err := c.do("", func(conn *iopool.Buffer) error {
		// write the header component
		if _, err := fmt.Fprintf(conn, "stats slabs\r\n"); err != nil {
			return err
		}

		// flush the connection, forcing bytes over the wire
		if err := conn.Flush(); err != nil {
			return err
		}

		// extract the slab stats payload
		payload, perr := slabs(conn.Reader)
		if perr != nil {
			return perr
		}
		statistics = payload

		return nil
	})

	return statistics, err
}

// StatsItems returns item statistics for a single memcached server.
//
// Note: this operation is performed on a single memcached server, even when
// the Client is configured with multiple server addresses. This is intentional,
// as stats is typically used by local monitoring tools that connect to a
// single memcached instance.
func StatsItems(c *Client) ([]*ItemStatistics, error) {
	var statistics []*ItemStatistics

	err := c.do("", func(conn *iopool.Buffer) error {
		// write the header component
		if _, err := fmt.Fprintf(conn, "stats items\r\n"); err != nil {
			return err
		}

		// flush the connection, forcing bytes over the wire
		if err := conn.Flush(); err != nil {
			return err
		}

		// extract the items stats payload
		payload, perr := items(conn.Reader)
		if perr != nil {
			return perr
		}
		statistics = payload

		return nil
	})

	return statistics, err
}

func unexpected(response []byte) error {
	return fmt.Errorf(
		"unexpected response from memcached %q",
		string(response),
	)
}