#include <stdio.h>

int main(void) {
    printf("hello from C\n");
    return 0;
}

#ifndef GREET_H
#define GREET_H

/* Declared here, defined in greet.c. Every file that includes this header can call it. */
void greet(const char *who);

#endif

#include <stdio.h>

#include "greet.h"

/* static: only this file can see the name. The linker never hears of it. */
static const char *prefix(void) {
    return "hello from";
}

/* No static: this name has external linkage. units.c calls it through greet.h. */
void greet(const char *who) {
    printf("%s %s\n", prefix(), who);
}

#include "greet.h"

int main(void) {
    greet("two files");
    return 0;
}

# Two translation units. Each .c is compiled alone into a .o, and the linker joins them.
units: units.o greet.o
	$(CC) $(CFLAGS) -o $@ units.o greet.o

units.o: units.c greet.h
	$(CC) $(CFLAGS) -c units.c

greet.o: greet.c greet.h
	$(CC) $(CFLAGS) -c greet.c

package main

import (
	"fmt"
	"runtime"
	"unsafe"
)

func main() {
	fmt.Println("hello from Go")
}

module hello

go 1.27

[package]
name = "hello"
version = "0.1.0"
edition = "2024"

[dependencies]

fn main() {
    println!("hello from Rust");
}

CC     ?= cc
CFLAGS ?= -std=c17 -Wall -Wextra -Werror -O2

hello: hello.c
	$(CC) $(CFLAGS) -o $@ hello.c

# Two translation units. Each .c is compiled alone into a .o, and the linker joins them.
units: units.o greet.o
	$(CC) $(CFLAGS) -o $@ units.o greet.o

units.o: units.c greet.h
	$(CC) $(CFLAGS) -c units.c

greet.o: greet.c greet.h
	$(CC) $(CFLAGS) -c greet.c

clean:
	rm -rf hello units *.o hello.dSYM units.dSYM

.PHONY: clean

/* The port comes from the environment, so the tests can pick a free one. */
static int kv_port(void) {
    const char *s = getenv("KV_PORT");
    if (s == NULL || *s == '\0') {
        return 6380;
    }
    return atoi(s);
}

/* Open a listening socket on every interface. Returns the socket, or -1 after
 * printing the reason. The caller owns the socket and must close() it. */
static int kv_listen(int port) {
    int fd = socket(AF_INET, SOCK_STREAM, 0);
    if (fd < 0) {
        perror("socket");
        return -1;
    }
    /* Without this, a restart within a minute fails with "address in use". */
    int yes = 1;
    if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof yes) < 0) {
        perror("setsockopt");
        goto cleanup;
    }
    struct sockaddr_in addr;
    memset(&addr, 0, sizeof addr);
    addr.sin_family = AF_INET;
    addr.sin_addr.s_addr = htonl(INADDR_ANY);
    addr.sin_port = htons((uint16_t)port);
    if (bind(fd, (struct sockaddr *)&addr, sizeof addr) < 0) {
        perror("bind");
        goto cleanup;
    }
    if (listen(fd, 16) < 0) {
        perror("listen");
        goto cleanup;
    }
    return fd;

cleanup:
    close(fd);
    return -1;
}

int main(void) {
    int port = kv_port();
    int fd = kv_listen(port);
    if (fd < 0) {
        return 1;
    }
    fprintf(stderr, "kv listening on port %d\n", port);
    for (;;) {
        int client = accept(fd, NULL, NULL);
        if (client < 0) {
            if (errno == EINTR) {
                continue;
            }
            perror("accept");
            break;
        }
        kv_serve(client);
        close(client);
    }
    close(fd);
    return 1;
}

CC      ?= cc
CFLAGS  ?= -std=c17 -Wall -Wextra -Werror -O2 -g
SANITIZE ?=

kv: kv.c
	$(CC) $(CFLAGS) $(SANITIZE) -o $@ kv.c

clean:
	rm -rf kv kv.dSYM

.PHONY: clean

// The port comes from the environment, so the tests can pick a free one.
func port() string {
	if p := os.Getenv("KV_PORT"); p != "" {
		return p
	}
	return "6380"
}

// Two rules of defer, as a test you can run with go test: the deferred calls run last in
// first out, and each one's arguments are evaluated at the defer line, not at the exit.
func deferOrder(log *[]string) {
	n := 1
	defer func() { *log = append(*log, "third, closure sees n =", itoa(n)) }()
	defer func(v int) { *log = append(*log, "second, argument was", itoa(v)) }(n)
	n = 2
	*log = append(*log, "first, the body")
}

func TestDeferOrder(t *testing.T) {
	var log []string
	deferOrder(&log)
	want := []string{"first, the body", "second, argument was", "1", "third, closure sees n =", "2"}
	for i := range want {
		if i >= len(log) || log[i] != want[i] {
			t.Fatalf("log = %q, want %q", log, want)
		}
	}
}

func main() {
	ln, err := net.Listen("tcp", ":"+port())
	if err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(1)
	}
	fmt.Fprintln(os.Stderr, "kv listening on port", port())
	for {
		conn, err := ln.Accept()
		if err != nil {
			fmt.Fprintln(os.Stderr, "accept:", err)
			continue
		}
		if err := serve(conn); err != nil {
			fmt.Fprintln(os.Stderr, "client:", err)
		}
	}
}

/// The port comes from the environment, so the tests can pick a free one.
fn port() -> String {
    std::env::var("KV_PORT").unwrap_or_else(|_| "6380".to_string())
}

fn main() -> io::Result<()> {
    let port = port();
    let listener = TcpListener::bind(format!("0.0.0.0:{port}"))?;
    eprintln!("kv listening on port {port}");
    for stream in listener.incoming() {
        match stream {
            Ok(stream) => {
                if let Err(e) = serve(stream) {
                    eprintln!("client: {e}");
                }
            }
            Err(e) => eprintln!("accept: {e}"),
        }
    }
    Ok(())
}

/* A growable byte buffer. It owns data, and kv_buf_free releases it. */
typedef struct {
    char *data; /* owned; NULL until the first append */
    size_t len;
    size_t cap;
} kv_buf;

static bool kv_buf_reserve(kv_buf *b, size_t extra) {
    if (b->len + extra <= b->cap) {
        return true;
    }
    size_t cap = b->cap ? b->cap * 2 : 1024;
    while (cap < b->len + extra) {
        cap *= 2;
    }
    char *p = realloc(b->data, cap);
    if (p == NULL) {
        return false; /* the old block is still valid and still ours */
    }
    b->data = p;
    b->cap = cap;
    return true;
}

static bool kv_buf_append(kv_buf *b, const void *src, size_t n) {
    if (!kv_buf_reserve(b, n)) {
        return false;
    }
    memcpy(b->data + b->len, src, n);
    b->len += n;
    return true;
}

/* Drop the first n bytes. memmove allows the two ranges to overlap. */
static void kv_buf_consume(kv_buf *b, size_t n) {
    memmove(b->data, b->data + n, b->len - n);
    b->len -= n;
}

static void kv_buf_free(kv_buf *b) {
    free(b->data);
    b->data = NULL;
    b->len = b->cap = 0;
}

/* One parsed request. The arguments point into the request buffer, so they are valid
 * only until that buffer is consumed. A kv_cmd owns nothing. */
typedef struct {
    size_t argc;
    const char *argv[KV_MAX_ARGS];
    size_t arglen[KV_MAX_ARGS];
} kv_cmd;

typedef enum {
    KV_PARSE_OK,         /* cmd is filled in, and *used bytes were taken */
    KV_PARSE_INCOMPLETE, /* read more bytes and try again */
    KV_PARSE_BAD         /* the bytes are not RESP: close the connection */
} kv_parse_status;

/* The '\n' that ends the line starting at p, or NULL when it has not arrived yet. */
static const char *kv_find_eol(const char *p, const char *end) {
    return memchr(p, '\n', (size_t)(end - p));
}

/* A decimal number between p and the line end. False when the text is not a number. */
static bool kv_parse_number(const char *p, const char *eol, long *out) {
    const char *stop = eol;
    if (stop > p && stop[-1] == '\r') {
        stop--;
    }
    if (p == stop || stop - p > 18) {
        return false;
    }
    long v = 0;
    for (; p < stop; p++) {
        if (*p < '0' || *p > '9') {
            return false;
        }
        v = v * 10 + (*p - '0');
    }
    *out = v;
    return true;
}

/* An inline command: words separated by spaces, as typed into nc or telnet. */
static kv_parse_status kv_parse_inline(const char *data, const char *eol, kv_cmd *cmd,
                                       size_t *used) {
    const char *p = data;
    const char *stop = eol;
    if (stop > p && stop[-1] == '\r') {
        stop--;
    }
    while (p < stop) {
        while (p < stop && *p == ' ') {
            p++;
        }
        if (p == stop) {
            break;
        }
        const char *word = p;
        while (p < stop && *p != ' ') {
            p++;
        }
        if (cmd->argc == KV_MAX_ARGS) {
            return KV_PARSE_BAD;
        }
        cmd->argv[cmd->argc] = word;
        cmd->arglen[cmd->argc] = (size_t)(p - word);
        cmd->argc++;
    }
    *used = (size_t)(eol + 1 - data);
    return KV_PARSE_OK;
}

static bool kv_reply_line(kv_buf *out, char first, const char *s) {
    return kv_buf_append(out, &first, 1) && kv_buf_append(out, s, strlen(s)) &&
           kv_buf_append(out, "\r\n", 2);
}

static bool kv_reply_simple(kv_buf *out, const char *s) { return kv_reply_line(out, '+', s); }
static bool kv_reply_error(kv_buf *out, const char *s) { return kv_reply_line(out, '-', s); }

static bool kv_reply_bulk(kv_buf *out, const char *s, size_t n) {
    char head[32];
    int h = snprintf(head, sizeof head, "$%zu\r\n", n);
    return kv_buf_append(out, head, (size_t)h) && kv_buf_append(out, s, n) &&
           kv_buf_append(out, "\r\n", 2);
}

static bool kv_reply_array(kv_buf *out, size_t n) {
    char head[32];
    int h = snprintf(head, sizeof head, "*%zu\r\n", n);
    return kv_buf_append(out, head, (size_t)h);
}

/* An error reply with a format, as printf would write it. The ... takes any arguments,
 * va_list walks them, and vsnprintf is the printf that accepts a va_list. */
static bool kv_reply_errorf(kv_buf *out, const char *fmt, ...) {
    char msg[128];
    va_list ap;
    va_start(ap, fmt);
    vsnprintf(msg, sizeof msg, fmt, ap);
    va_end(ap);
    return kv_reply_error(out, msg);
}

/* True when argument i of the command is name, ignoring case. */
static bool kv_arg_is(const kv_cmd *c, size_t i, const char *name) {
    size_t n = strlen(name);
    return c->arglen[i] == n && strncasecmp(c->argv[i], name, n) == 0;
}

static bool kv_reply_arity(kv_buf *out, const char *name) {
    return kv_reply_errorf(out, "ERR wrong number of arguments for '%s' command", name);
}

/* Run one command and append its reply to out. */
static bool kv_exec(const kv_cmd *c, kv_buf *out) {
    if (c->argc == 0) {
        return kv_reply_error(out, "ERR empty command");
    }
    if (kv_arg_is(c, 0, "PING")) {
        if (c->argc == 1) {
            return kv_reply_simple(out, "PONG");
        }
        if (c->argc == 2) {
            return kv_reply_bulk(out, c->argv[1], c->arglen[1]);
        }
        return kv_reply_arity(out, "ping");
    }
    if (kv_arg_is(c, 0, "ECHO")) {
        if (c->argc != 2) {
            return kv_reply_arity(out, "echo");
        }
        return kv_reply_bulk(out, c->argv[1], c->arglen[1]);
    }
    if (kv_arg_is(c, 0, "COMMAND")) {
        return kv_reply_array(out, 0); /* redis-cli asks on connect; an empty list is fine */
    }
    int n = c->arglen[0] < 40 ? (int)c->arglen[0] : 40;
    return kv_reply_errorf(out, "ERR unknown command '%.*s'", n, c->argv[0]);
}

/* write() may send fewer bytes than asked, so loop until all are out. */
static bool kv_write_all(int fd, const char *p, size_t n) {
    size_t off = 0;
    while (off < n) {
        ssize_t w = write(fd, p + off, n - off);
        if (w < 0) {
            if (errno == EINTR) {
                continue;
            }
            perror("write");
            return false;
        }
        off += (size_t)w;
    }
    return true;
}

/* Read requests, run them, send the replies, until the client closes. */
static void kv_serve(int client) {
    kv_buf in = {0};
    kv_buf out = {0};
    char tmp[4096];
    for (;;) {
        ssize_t n = read(client, tmp, sizeof tmp);
        if (n < 0) {
            if (errno == EINTR) {
                continue;
            }
            perror("read");
            break;
        }
        if (n == 0) {
            break; /* the client closed the connection */
        }
        if (!kv_buf_append(&in, tmp, (size_t)n)) {
            break;
        }
        /* A pipeline holds several requests in one read, so loop until one is incomplete. */
        bool bad = false;
        for (;;) {
            kv_cmd cmd;
            size_t used = 0;
            kv_parse_status st = kv_parse(in.data, in.len, &cmd, &used);
            /* No default: with -Wall the compiler reports a status this switch forgot. */
            switch (st) {
            case KV_PARSE_INCOMPLETE:
                bad = in.len > KV_MAX_REQUEST; /* a request that never ends is not a request */
                break;
            case KV_PARSE_BAD:
                bad = true;
                break;
            case KV_PARSE_OK:
                if (!kv_exec(&cmd, &out)) {
                    bad = true;
                }
                kv_buf_consume(&in, used);
                continue;
            }
            break;
        }
        if (bad) {
            kv_reply_error(&out, "ERR Protocol error");
        }
        if (out.len > 0 && !kv_write_all(client, out.data, out.len)) {
            break;
        }
        out.len = 0;
        if (bad) {
            break;
        }
    }
    kv_buf_free(&in);
    kv_buf_free(&out);
}

// readLine returns one line without its line ending. A bare LF is accepted as well as
// CRLF, so that a command typed into nc works. io.EOF means the client closed the
// connection cleanly between commands.
func readLine(r *bufio.Reader) ([]byte, error) {
	line, err := r.ReadBytes('\n')
	if err != nil {
		if err == io.EOF && len(line) > 0 {
			return nil, io.ErrUnexpectedEOF
		}
		return nil, err
	}
	line = bytes.TrimSuffix(line, []byte("\n"))
	return bytes.TrimSuffix(line, []byte("\r")), nil
}

// errProtocol means the bytes were not RESP. The connection is closed after it. A caller
// that only needs to know that much tests errors.Is(err, errProtocol).
var errProtocol = errors.New("protocol error")

// A ProtocolError says which byte was wrong. Unwrap returns the sentinel, so errors.Is still
// finds errProtocol in it, and errors.As gives a caller the byte.
type ProtocolError struct {
	Byte byte
}

func (e *ProtocolError) Error() string { return fmt.Sprintf("protocol error at byte %q", e.Byte) }
func (e *ProtocolError) Unwrap() error { return errProtocol }

func protoErr(b byte) error { return &ProtocolError{Byte: b} }

func writeSimple(w *bufio.Writer, s string) error {
	_, err := w.WriteString("+" + s + "\r\n")
	return err
}

func writeError(w *bufio.Writer, s string) error {
	_, err := w.WriteString("-" + s + "\r\n")
	return err
}

func writeBulk(w *bufio.Writer, b []byte) error {
	if _, err := w.WriteString("$" + strconv.Itoa(len(b)) + "\r\n"); err != nil {
		return err
	}
	if _, err := w.Write(b); err != nil {
		return err
	}
	_, err := w.WriteString("\r\n")
	return err
}

func writeArrayLen(w *bufio.Writer, n int) error {
	_, err := w.WriteString("*" + strconv.Itoa(n) + "\r\n")
	return err
}

// execute runs one command and writes its reply.
func execute(args [][]byte, w *bufio.Writer) error {
	if len(args) == 0 {
		return writeError(w, "ERR empty command")
	}
	name := strings.ToUpper(string(args[0]))
	switch name {
	case "PING":
		switch len(args) {
		case 1:
			return writeSimple(w, "PONG")
		case 2:
			return writeBulk(w, args[1])
		}
		return writeArity(w, "ping")
	case "ECHO":
		if len(args) != 2 {
			return writeArity(w, "echo")
		}
		return writeBulk(w, args[1])
	case "COMMAND":
		return writeArrayLen(w, 0) // redis-cli asks on connect; an empty list is fine
	}
	shown := string(args[0])
	if len(shown) > 40 {
		shown = shown[:40]
	}
	return writeError(w, "ERR unknown command '"+shown+"'")
}

func writeArity(w *bufio.Writer, name string) error {
	return writeError(w, "ERR wrong number of arguments for '"+name+"' command")
}

// serve reads requests, runs them and sends the replies, until the client closes.
func serve(conn net.Conn) error {
	defer conn.Close()
	r := bufio.NewReader(conn)
	w := bufio.NewWriter(conn)
	for {
		args, err := readCommand(r)
		// The error is wrapped, so == would miss io.EOF. errors.Is walks the chain.
		if errors.Is(err, io.EOF) {
			return nil
		}
		// errors.AsType finds the typed error in the chain and gives us its byte. Before
		// Go 1.26 the same check was errors.As(err, &pe) with a declared *ProtocolError.
		if pe, ok := errors.AsType[*ProtocolError](err); ok {
			writeError(w, "ERR Protocol error")
			w.Flush()
			return fmt.Errorf("closing client: %w", pe)
		}
		if err != nil {
			return err
		}
		if err := execute(args, w); err != nil {
			return err
		}
		// A pipeline holds several requests; flush once the reader has nothing waiting.
		if r.Buffered() == 0 {
			if err := w.Flush(); err != nil {
				return err
			}
		}
	}
}

/// One line without its line ending. A bare LF is accepted as well as CRLF, so that a
/// command typed into nc works. `Ok(None)` means the client closed the connection
/// cleanly between commands.
fn read_line(r: &mut impl BufRead) -> Result<Option<Vec<u8>>> {
    let mut line = Vec::new();
    if r.read_until(b'\n', &mut line)? == 0 {
        return Ok(None);
    }
    if line.pop() != Some(b'\n') {
        return Err(io::Error::from(io::ErrorKind::UnexpectedEof).into());
    }
    if line.last() == Some(&b'\r') {
        line.pop();
    }
    Ok(Some(line))
}

fn parse_len(digits: &[u8], type_byte: u8) -> Result<usize> {
    std::str::from_utf8(digits)
        .ok()
        .and_then(|s| s.parse().ok())
        .ok_or(KvError::Protocol(type_byte))
}

/// Everything that can go wrong while reading a request. Each variant carries what the
/// caller needs to know: the io error itself, or the byte that was not RESP.
#[derive(Debug)]
pub enum KvError {
    Io(io::Error),
    Protocol(u8),
}

/// How the error prints, with `{}`. `to_string` and `eprintln!` use this.
impl fmt::Display for KvError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            KvError::Io(e) => write!(f, "io error: {e}"),
            KvError::Protocol(b) => write!(f, "protocol error at byte {:?}", *b as char),
        }
    }
}

/// Joining the standard error trait lets a `KvError` sit in a `Box<dyn Error>`.
impl std::error::Error for KvError {}

/// `?` calls this when an `io::Error` must become a `KvError`. Without it, every read in
/// this file would need `.map_err(KvError::Io)`.
impl From<io::Error> for KvError {
    fn from(e: io::Error) -> Self {
        KvError::Io(e)
    }
}

/// The other direction, so a function that returns `io::Result` can use `?` on ours.
impl From<KvError> for io::Error {
    fn from(e: KvError) -> Self {
        match e {
            KvError::Io(e) => e,
            KvError::Protocol(_) => io::Error::new(io::ErrorKind::InvalidData, e.to_string()),
        }
    }
}

pub type Result<T> = std::result::Result<T, KvError>;

pub fn write_simple(w: &mut impl Write, s: &str) -> io::Result<()> {
    write!(w, "+{s}\r\n")
}

pub fn write_error(w: &mut impl Write, s: &str) -> io::Result<()> {
    write!(w, "-{s}\r\n")
}

pub fn write_bulk(w: &mut impl Write, b: &[u8]) -> io::Result<()> {
    write!(w, "${}\r\n", b.len())?;
    w.write_all(b)?;
    w.write_all(b"\r\n")
}

pub fn write_array_len(w: &mut impl Write, n: usize) -> io::Result<()> {
    write!(w, "*{n}\r\n")
}

/// Run one command and write its reply.
pub fn execute(args: &[Vec<u8>], w: &mut impl Write) -> io::Result<()> {
    let Some(name) = args.first() else {
        return resp::write_error(w, "ERR empty command");
    };
    match name.to_ascii_uppercase().as_slice() {
        b"PING" => match args.len() {
            1 => resp::write_simple(w, "PONG"),
            2 => resp::write_bulk(w, &args[1]),
            _ => write_arity(w, "ping"),
        },
        b"ECHO" => match args.len() {
            2 => resp::write_bulk(w, &args[1]),
            _ => write_arity(w, "echo"),
        },
        // redis-cli asks on connect; an empty list is fine
        b"COMMAND" => resp::write_array_len(w, 0),
        _ => {
            let shown = String::from_utf8_lossy(&name[..name.len().min(40)]);
            resp::write_error(w, &format!("ERR unknown command '{shown}'"))
        }
    }
}

fn write_arity(w: &mut impl Write, name: &str) -> io::Result<()> {
    resp::write_error(
        w,
        &format!("ERR wrong number of arguments for '{name}' command"),
    )
}

/// Read requests, run them and send the replies, until the client closes. The function
/// returns a `KvError`, so `?` converts every `io::Error` on the way with `From`.
fn serve(stream: TcpStream) -> resp::Result<()> {
    let mut reader = BufReader::new(stream.try_clone()?);
    let mut writer = BufWriter::new(stream);
    loop {
        match resp::read_command(&mut reader) {
            Ok(Some(args)) => cmd::execute(&args, &mut writer)?,
            Ok(None) => return Ok(()),
            Err(e @ KvError::Protocol(_)) => {
                resp::write_error(&mut writer, "ERR Protocol error")?;
                writer.flush()?;
                return Err(e);
            }
            Err(e) => return Err(e),
        }
        // A pipeline holds several requests; flush once the reader has nothing waiting.
        if reader.buffer().is_empty() {
            writer.flush()?;
        }
    }
}

/* A copy of n bytes that the caller owns and must free. Never NULL on success, even
 * for n == 0, because malloc(0) may return NULL and that would look like failure. */
static char *kv_bytes_dup(const char *src, size_t n) {
    char *p = malloc(n ? n : 1);
    if (p == NULL) {
        return NULL;
    }
    memcpy(p, src, n);
    return p;
}

static bool kv_bytes_eq(const char *a, size_t alen, const char *b, size_t blen) {
    return alen == blen && memcmp(a, b, alen) == 0;
}

/* Parse a signed decimal integer from bytes that have no terminating zero. Strict, as
 * Redis is: no spaces, no plus sign, no leading zeros, and no overflow. */
static bool kv_parse_int(const char *s, size_t n, long long *out) {
    size_t i = 0;
    bool neg = false;
    if (n == 0) {
        return false;
    }
    if (s[0] == '-') {
        neg = true;
        i = 1;
        if (n == 1) {
            return false;
        }
    }
    if (s[i] == '0' && n - i > 1) {
        return false;
    }
    unsigned long long v = 0;
    for (; i < n; i++) {
        if (s[i] < '0' || s[i] > '9') {
            return false;
        }
        unsigned d = (unsigned)(s[i] - '0');
        if (v > (ULLONG_MAX - d) / 10) {
            return false;
        }
        v = v * 10 + d;
    }
    if (!neg && v > (unsigned long long)LLONG_MAX) {
        return false;
    }
    if (neg && v > (unsigned long long)LLONG_MAX + 1) {
        return false;
    }
    *out = neg ? -(long long)(v - 1) - 1 : (long long)v;
    return true;
}

/* One key and its value. Both are owned copies, freed with the entry. */
typedef struct {
    char *key;
    size_t klen;
    char *val;
    size_t vlen;
} kv_entry;

/* Every entry, in a list. A lookup walks the whole list, which chapter 6 fixes. */
typedef struct {
    kv_entry *items; /* owned */
    size_t len;
    size_t cap;
} kv_store;

static kv_entry *kv_store_find(kv_store *s, const char *key, size_t klen) {
    for (size_t i = 0; i < s->len; i++) {
        if (kv_bytes_eq(s->items[i].key, s->items[i].klen, key, klen)) {
            return &s->items[i];
        }
    }
    return NULL;
}

/* INCR: the value is bytes that must parse as an integer, and the new value is bytes too.
 * The check against LLONG_MAX comes before the add, because overflowing a signed integer
 * is undefined behaviour, and the compiler may assume it never happens. */
static bool kv_cmd_incr(const kv_cmd *c, kv_store *s, kv_buf *out) {
    long long n = 0;
    kv_entry *e = kv_store_find(s, c->argv[1], c->arglen[1]);
    if (e != NULL && !kv_parse_int(e->val, e->vlen, &n)) {
        return kv_reply_error(out, "ERR value is not an integer or out of range");
    }
    if (n == LLONG_MAX) {
        return kv_reply_error(out, "ERR increment or decrement would overflow");
    }
    n++;
    char text[32];
    int len = snprintf(text, sizeof text, "%lld", n);
    if (!kv_store_set(s, c->argv[1], c->arglen[1], text, (size_t)len)) {
        return false;
    }
    return kv_reply_int(out, n);
}

/* Each command is a function of this one shape, so a table can hold them all. */
typedef bool (*kv_cmd_fn)(const kv_cmd *c, kv_store *s, kv_buf *out);

static bool kv_cmd_ping(const kv_cmd *c, kv_store *s, kv_buf *out) {
    (void)s;
    if (c->argc == 1) {
        return kv_reply_simple(out, "PONG");
    }
    if (c->argc == 2) {
        return kv_reply_bulk(out, c->argv[1], c->arglen[1]);
    }
    return kv_reply_arity(out, "ping");
}

static bool kv_cmd_echo(const kv_cmd *c, kv_store *s, kv_buf *out) {
    (void)s;
    return kv_reply_bulk(out, c->argv[1], c->arglen[1]);
}

static bool kv_cmd_set(const kv_cmd *c, kv_store *s, kv_buf *out) {
    if (!kv_store_set(s, c->argv[1], c->arglen[1], c->argv[2], c->arglen[2])) {
        return false;
    }
    return kv_reply_simple(out, "OK");
}

static bool kv_cmd_get(const kv_cmd *c, kv_store *s, kv_buf *out) {
    kv_entry *e = kv_store_find(s, c->argv[1], c->arglen[1]);
    return e ? kv_reply_bulk(out, e->val, e->vlen) : kv_reply_nil(out);
}

static bool kv_cmd_del(const kv_cmd *c, kv_store *s, kv_buf *out) {
    long long n = 0;
    for (size_t i = 1; i < c->argc; i++) {
        n += kv_store_del(s, c->argv[i], c->arglen[i]);
    }
    return kv_reply_int(out, n);
}

static bool kv_cmd_exists(const kv_cmd *c, kv_store *s, kv_buf *out) {
    long long n = 0;
    for (size_t i = 1; i < c->argc; i++) {
        n += kv_store_find(s, c->argv[i], c->arglen[i]) != NULL;
    }
    return kv_reply_int(out, n);
}

static bool kv_cmd_command(const kv_cmd *c, kv_store *s, kv_buf *out) {
    (void)c;
    (void)s;
    return kv_reply_array(out, 0); /* redis-cli asks on connect; an empty list is fine */
}

/* The command list, written once. Each use of KV_COMMANDS defines X to say what to make
 * of a row. The first use below builds the table, the second counts it. A name is spelled
 * as the arity error must show it, and arity is counted as Redis counts it: a positive
 * number is the exact argc, name included, and a negative one is a minimum. */
#define KV_COMMANDS                     \
    X(ping, kv_cmd_ping, -1)            \
    X(echo, kv_cmd_echo, 2)             \
    X(set, kv_cmd_set, 3)               \
    X(get, kv_cmd_get, 2)               \
    X(del, kv_cmd_del, -2)              \
    X(exists, kv_cmd_exists, -2)        \
    X(incr, kv_cmd_incr, 2)             \
    X(append, kv_cmd_append, 3)         \
    X(strlen, kv_cmd_strlen, 2)         \
    X(getrange, kv_cmd_getrange, 4)     \
    X(command, kv_cmd_command, -1)

typedef struct {
    const char *name;
    kv_cmd_fn fn;
    int arity;
} kv_command;

static const kv_command kv_commands[] = {
#define X(name, fn, arity) {#name, fn, arity},
    KV_COMMANDS
#undef X
};

enum {
#define X(name, fn, arity) +1
    KV_COMMAND_COUNT = 0 KV_COMMANDS
#undef X
};

/* Run one command against the store and append its reply to out: find the row, check
 * the arity, call the function through its pointer. */
static bool kv_exec(const kv_cmd *c, kv_store *s, kv_buf *out) {
    if (c->argc == 0) {
        return kv_reply_error(out, "ERR empty command");
    }
    for (size_t i = 0; i < KV_COMMAND_COUNT; i++) {
        const kv_command *cmd = &kv_commands[i];
        if (!kv_arg_is(c, 0, cmd->name)) {
            continue;
        }
        int argc = (int)c->argc;
        bool ok = cmd->arity < 0 ? argc >= -cmd->arity : argc == cmd->arity;
        if (!ok) {
            return kv_reply_arity(out, cmd->name);
        }
        return cmd->fn(c, s, out);
    }
    int n = c->arglen[0] < 40 ? (int)c->arglen[0] : 40;
    return kv_reply_errorf(out, "ERR unknown command '%.*s'", n, c->argv[0]);
}

/* APPEND key value: grow the stored bytes in place. A missing key starts empty. */
static bool kv_cmd_append(const kv_cmd *c, kv_store *s, kv_buf *out) {
    kv_entry *e = kv_store_find(s, c->argv[1], c->arglen[1]);
    if (e == NULL) {
        if (!kv_store_set(s, c->argv[1], c->arglen[1], c->argv[2], c->arglen[2])) {
            return false;
        }
        return kv_reply_int(out, (long long)c->arglen[2]);
    }
    char *v = realloc(e->val, e->vlen + c->arglen[2] + 1);
    if (v == NULL) {
        return false; /* e->val is still valid: realloc leaves it alone on failure */
    }
    memcpy(v + e->vlen, c->argv[2], c->arglen[2]);
    e->val = v;
    e->vlen += c->arglen[2];
    return kv_reply_int(out, (long long)e->vlen);
}

/* STRLEN key: the length in bytes, which is the only length the store knows. */
static bool kv_cmd_strlen(const kv_cmd *c, kv_store *s, kv_buf *out) {
    kv_entry *e = kv_store_find(s, c->argv[1], c->arglen[1]);
    return kv_reply_int(out, e ? (long long)e->vlen : 0);
}

/* GETRANGE key start end: a slice by byte offsets, both inclusive. A negative offset
 * counts from the end, and an offset past either end is clamped, as Redis does. */
static bool kv_cmd_getrange(const kv_cmd *c, kv_store *s, kv_buf *out) {
    long long start, end;
    if (!kv_parse_int(c->argv[2], c->arglen[2], &start) ||
        !kv_parse_int(c->argv[3], c->arglen[3], &end)) {
        return kv_reply_error(out, "ERR value is not an integer or out of range");
    }
    kv_entry *e = kv_store_find(s, c->argv[1], c->arglen[1]);
    long long len = e ? (long long)e->vlen : 0;
    if (start < 0) {
        start += len;
    }
    if (end < 0) {
        end += len;
    }
    if (start < 0) {
        start = 0;
    }
    if (end >= len) {
        end = len - 1;
    }
    if (len == 0 || start > end) {
        return kv_reply_bulk(out, "", 0);
    }
    return kv_reply_bulk(out, e->val + start, (size_t)(end - start + 1));
}

// store keeps every key and its value. The parser allocated each slice, so the store can
// keep them without copying: nothing else holds them.
type store struct {
	m map[string][]byte
}

func newStore() *store {
	return &store{m: make(map[string][]byte)}
}

func (s *store) get(key []byte) ([]byte, bool) {
	v, ok := s.m[string(key)]
	return v, ok
}

func (s *store) set(key, val []byte) {
	s.m[string(key)] = val
}

func (s *store) del(key []byte) bool {
	_, ok := s.m[string(key)]
	delete(s.m, string(key))
	return ok
}

func (s *store) has(key []byte) bool {
	_, ok := s.m[string(key)]
	return ok
}

// The two ways INCR fails. Their text is what the client sees.
var (
	errNotInteger = errors.New("ERR value is not an integer or out of range")
	errOverflow   = errors.New("ERR increment or decrement would overflow")
)

// A handler is any function of this shape. The table maps the upper-cased name to one,
// with the arity as Redis counts it: a positive number is the exact argc, name included,
// and a negative one is a minimum.
type handler func(args [][]byte, s *store, w *bufio.Writer) error

type command struct {
	run   handler
	arity int
}

var commands = map[string]command{
	"PING":     {ping, -1},
	"ECHO":     {echo, 2},
	"SET":      {set, 3},
	"GET":      {get, 2},
	"DEL":      {del, -2},
	"EXISTS":   {exists, -2},
	"INCR":     {incr, 2},
	"APPEND":   {appendCmd, 3},
	"STRLEN":   {strlenCmd, 2},
	"GETRANGE": {getrange, 4},
	"COMMAND":  {commandCmd, -1},
}

// execute runs one command against the store and writes its reply: look the name up,
// check the arity, call the function the table holds.
func execute(args [][]byte, s *store, w *bufio.Writer) error {
	if len(args) == 0 {
		return writeError(w, "ERR empty command")
	}
	name := strings.ToUpper(string(args[0]))
	cmd, ok := commands[name]
	if !ok {
		shown := string(args[0])
		if len(shown) > 40 {
			shown = shown[:40]
		}
		return writeError(w, "ERR unknown command '"+shown+"'")
	}
	if n := len(args); (cmd.arity < 0 && n < -cmd.arity) || (cmd.arity > 0 && n != cmd.arity) {
		return writeArity(w, strings.ToLower(name))
	}
	return cmd.run(args, s, w)
}

func writeArity(w *bufio.Writer, name string) error {
	return writeError(w, "ERR wrong number of arguments for '"+name+"' command")
}

func ping(args [][]byte, _ *store, w *bufio.Writer) error {
	switch len(args) {
	case 1:
		return writeSimple(w, "PONG")
	case 2:
		return writeBulk(w, args[1])
	}
	return writeArity(w, "ping")
}

func echo(args [][]byte, _ *store, w *bufio.Writer) error {
	return writeBulk(w, args[1])
}

func set(args [][]byte, s *store, w *bufio.Writer) error {
	s.set(args[1], args[2])
	return writeSimple(w, "OK")
}

func get(args [][]byte, s *store, w *bufio.Writer) error {
	if v, ok := s.get(args[1]); ok {
		return writeBulk(w, v)
	}
	return writeNil(w)
}

// countKeys counts the keys for which hit returns true. DEL and EXISTS differ only in
// which method they pass: s.del removes as it counts, s.has only looks.
func countKeys(keys [][]byte, hit func([]byte) bool) int64 {
	var n int64
	for _, key := range keys {
		if hit(key) {
			n++
		}
	}
	return n
}

func del(args [][]byte, s *store, w *bufio.Writer) error {
	return writeInt(w, countKeys(args[1:], s.del))
}

func exists(args [][]byte, s *store, w *bufio.Writer) error {
	return writeInt(w, countKeys(args[1:], s.has))
}

func incr(args [][]byte, s *store, w *bufio.Writer) error {
	n, err := s.incr(args[1])
	if err != nil {
		return writeError(w, err.Error())
	}
	return writeInt(w, n)
}

func commandCmd(_ [][]byte, _ *store, w *bufio.Writer) error {
	return writeArrayLen(w, 0) // redis-cli asks on connect; an empty list is fine
}

// A string is bytes. len counts them, and a character outside ASCII takes more than one.
func TestBytesAndRunes(t *testing.T) {
	s := "héllo"
	if len(s) != 6 || utf8.RuneCountInString(s) != 5 {
		t.Fatalf("len %d, runes %d", len(s), utf8.RuneCountInString(s))
	}
	// range walks runes, not bytes: the index jumps from 1 to 3 over the two bytes of é
	var idx []int
	for i := range s {
		idx = append(idx, i)
	}
	if idx[2] != 3 {
		t.Fatalf("indexes %v", idx)
	}
	// s[1:2] is one byte, half of é, and not a valid string on its own
	if utf8.ValidString(s[1:2]) {
		t.Fatal("half a rune should not be valid")
	}
	// strings.Builder grows a string without copying it at every step
	var b strings.Builder
	for i := 0; i < 3; i++ {
		b.WriteString("ab")
	}
	if b.String() != "ababab" {
		t.Fatal(b.String())
	}
}

// appendCmd grows the stored bytes. append may reuse the old slice's spare capacity or
// allocate a new array; either way the map is updated with the slice it returns.
func appendCmd(args [][]byte, s *store, w *bufio.Writer) error {
	old, _ := s.get(args[1])
	v := append(old, args[2]...)
	s.set(args[1], v)
	return writeInt(w, int64(len(v)))
}

// strlenCmd replies with the length in bytes. len counts bytes, and so does Redis; the
// number of characters is a different question, which utf8.RuneCount answers.
func strlenCmd(args [][]byte, s *store, w *bufio.Writer) error {
	v, _ := s.get(args[1])
	_ = utf8.RuneCount(v) // the count a user would expect for text, which STRLEN is not
	return writeInt(w, int64(len(v)))
}

// getrange slices by byte offsets, both inclusive. A negative offset counts from the end,
// and an offset past either end is clamped, as Redis does.
func getrange(args [][]byte, s *store, w *bufio.Writer) error {
	start, err1 := strconv.ParseInt(string(args[2]), 10, 64)
	end, err2 := strconv.ParseInt(string(args[3]), 10, 64)
	if err1 != nil || err2 != nil {
		return writeError(w, "ERR value is not an integer or out of range")
	}
	v, _ := s.get(args[1])
	n := int64(len(v))
	if start < 0 {
		start += n
	}
	if end < 0 {
		end += n
	}
	start = max(start, 0)
	end = min(end, n-1)
	if n == 0 || start > end {
		return writeBulk(w, nil)
	}
	return writeBulk(w, v[start:end+1])
}

/// Every key and its value. The store owns both: a key or value that is put in is moved,
/// not copied, and is freed when it is replaced, deleted or the store is dropped.
#[derive(Default)]
pub struct Store {
    map: HashMap<Vec<u8>, Vec<u8>>,
}

impl Store {
    pub fn new() -> Self {
        Self::default()
    }

    /// A borrowed view of the value, which lives as long as the borrow of the store.
    pub fn get(&self, key: &[u8]) -> Option<&[u8]> {
        self.map.get(key).map(Vec::as_slice)
    }

    pub fn set(&mut self, key: Vec<u8>, val: Vec<u8>) {
        self.map.insert(key, val);
    }

    pub fn del(&mut self, key: &[u8]) -> bool {
        self.map.remove(key).is_some()
    }

    /// Append to the value under key, which starts empty, and return the new length. The
    /// entry API finds or inserts in one lookup, and `extend_from_slice` grows the `Vec`
    /// in place when it has room.
    pub fn append(&mut self, key: Vec<u8>, val: Vec<u8>) -> usize {
        let v = self.map.entry(key).or_default();
        v.extend_from_slice(&val);
        v.len()
    }
}

/// Every command is a function of this one shape, so a table can hold them. The writer is
/// `dyn Write`, any writer decided at run time, because a table cannot hold one function
/// per writer type. Chapter 7 explains `dyn`.
type Handler = fn(&mut Command, &mut Store, &mut dyn Write) -> io::Result<()>;

/// A row of the table: the name as the client spells it, the function, and the arity as
/// Redis counts it. Positive is the exact argc, name included; negative is a minimum.
struct Entry {
    name: &'static [u8],
    run: Handler,
    arity: i32,
}

/// The table, written once. The macro turns each `name => function, arity` line into an
/// `Entry`, so the list reads like a list and not like a struct literal eleven times.
macro_rules! commands {
    ($($name:literal => $run:ident, $arity:expr;)*) => {
        const COMMANDS: &[Entry] = &[$(Entry { name: $name, run: $run, arity: $arity }),*];
    };
}

commands! {
    b"PING" => ping, -1;
    b"ECHO" => echo, 2;
    b"SET" => set, 3;
    b"GET" => get, 2;
    b"DEL" => del, -2;
    b"EXISTS" => exists, -2;
    b"INCR" => incr, 2;
    b"APPEND" => append, 3;
    b"STRLEN" => strlen, 2;
    b"GETRANGE" => getrange, 4;
    b"COMMAND" => command, -1;
}

/// Run one command against the store and write its reply: find the row, check the arity,
/// call the function. The command is taken by value, so that SET can move the key and
/// the value into the store instead of copying them.
pub fn execute(mut args: Command, store: &mut Store, w: &mut impl Write) -> io::Result<()> {
    let Some(name) = args.first() else {
        return resp::write_error(w, "ERR empty command");
    };
    let name = name.to_ascii_uppercase();
    let Some(entry) = COMMANDS.iter().find(|e| e.name == name.as_slice()) else {
        let shown = String::from_utf8_lossy(&args[0][..args[0].len().min(40)]);
        return resp::write_error(w, &format!("ERR unknown command '{shown}'"));
    };
    let argc = args.len() as i32;
    let ok = if entry.arity < 0 {
        argc >= -entry.arity
    } else {
        argc == entry.arity
    };
    if !ok {
        return write_arity(w, entry.name);
    }
    (entry.run)(&mut args, store, w)
}

fn write_arity(w: &mut dyn Write, name: &[u8]) -> io::Result<()> {
    let name = String::from_utf8_lossy(name).to_ascii_lowercase();
    resp::write_error(
        w,
        &format!("ERR wrong number of arguments for '{name}' command"),
    )
}

fn ping(args: &mut Command, _: &mut Store, w: &mut dyn Write) -> io::Result<()> {
    match args.as_slice() {
        [_] => resp::write_simple(w, "PONG"),
        [_, msg] => resp::write_bulk(w, msg),
        _ => write_arity(w, b"ping"),
    }
}

fn echo(args: &mut Command, _: &mut Store, w: &mut dyn Write) -> io::Result<()> {
    resp::write_bulk(w, &args[1])
}

fn set(args: &mut Command, store: &mut Store, w: &mut dyn Write) -> io::Result<()> {
    let [_, key, val] = args.as_mut_slice() else {
        unreachable!("arity checked")
    };
    store.set(mem::take(key), mem::take(val));
    resp::write_simple(w, "OK")
}

fn get(args: &mut Command, store: &mut Store, w: &mut dyn Write) -> io::Result<()> {
    match store.get(&args[1]) {
        Some(val) => resp::write_bulk(w, val),
        None => resp::write_nil(w),
    }
}

fn del(args: &mut Command, store: &mut Store, w: &mut dyn Write) -> io::Result<()> {
    let n = args[1..].iter().filter(|k| store.del(k)).count();
    resp::write_int(w, n as i64)
}

fn exists(args: &mut Command, store: &mut Store, w: &mut dyn Write) -> io::Result<()> {
    let n = args[1..].iter().filter(|k| store.get(k).is_some()).count();
    resp::write_int(w, n as i64)
}

fn incr(args: &mut Command, store: &mut Store, w: &mut dyn Write) -> io::Result<()> {
    match store.incr(&args[1]) {
        Ok(n) => resp::write_int(w, n),
        Err(e) => resp::write_error(w, e.message()),
    }
}

fn command(_: &mut Command, _: &mut Store, w: &mut dyn Write) -> io::Result<()> {
    resp::write_array_len(w, 0) // redis-cli asks on connect; an empty list is fine
}

/// A `str` is bytes that are valid UTF-8. `len` counts the bytes, `chars` the
/// characters, and a slice must land on a character boundary.
#[test]
fn bytes_and_chars() {
    let s = "héllo";
    assert_eq!(s.len(), 6);
    assert_eq!(s.chars().count(), 5);
    assert_eq!(&s[0..1], "h");
    assert_eq!(&s[1..3], "é");
    assert!(s.is_char_boundary(3));
    assert!(!s.is_char_boundary(2));
    // the same bytes as a slice can be cut anywhere, which is what GETRANGE needs
    assert_eq!(&s.as_bytes()[1..2], &[0xC3]);
}

/// Slicing a `str` in the middle of a character is a panic, not a corrupt string.
#[test]
#[should_panic(expected = "byte index 2 is not a char boundary")]
fn half_a_char_panics() {
    let s = "héllo";
    let _ = &s[0..2];
}

/// APPEND key value: grow the stored bytes. The value is moved in when the key is new.
fn append(args: &mut Command, store: &mut Store, w: &mut dyn Write) -> io::Result<()> {
    let [_, key, val] = args.as_mut_slice() else {
        unreachable!("arity checked")
    };
    let n = store.append(mem::take(key), mem::take(val));
    resp::write_int(w, n as i64)
}

/// STRLEN key: the length in bytes. `len` on a `Vec<u8>` counts bytes, and so does Redis.
/// The number of characters is a different question, which `chars().count()` answers on
/// a `str`, and which the store cannot ask, because a value need not be text.
fn strlen(args: &mut Command, store: &mut Store, w: &mut dyn Write) -> io::Result<()> {
    let n = store.get(&args[1]).map_or(0, <[u8]>::len);
    resp::write_int(w, n as i64)
}

/// GETRANGE key start end: a slice by byte offsets, both inclusive. A negative offset
/// counts from the end, and an offset past either end is clamped, as Redis does.
fn getrange(args: &mut Command, store: &mut Store, w: &mut dyn Write) -> io::Result<()> {
    let (Some(start), Some(end)) = (parse_int(&args[2]), parse_int(&args[3])) else {
        return resp::write_error(w, "ERR value is not an integer or out of range");
    };
    let val = store.get(&args[1]).unwrap_or(b"");
    let len = val.len() as i64;
    let start = if start < 0 { start + len } else { start }.max(0);
    let end = if end < 0 { end + len } else { end }.min(len - 1);
    if len == 0 || start > end {
        return resp::write_bulk(w, b"");
    }
    resp::write_bulk(w, &val[start as usize..=end as usize])
}

/// Why INCR could not add one.
pub enum IncrError {
    NotInteger,
    Overflow,
}

impl IncrError {
    pub fn message(&self) -> &'static str {
        match self {
            Self::NotInteger => "ERR value is not an integer or out of range",
            Self::Overflow => "ERR increment or decrement would overflow",
        }
    }
}
/// The bytes as a signed 64-bit integer, if that is what they are.
pub fn parse_int(bytes: &[u8]) -> Option<i64> {
    std::str::from_utf8(bytes).ok()?.parse().ok()
}

// Go integers wrap. The largest int64 plus one is the smallest, with no error and no panic,
// which is why incr checks against math.MaxInt64 before it adds.
func TestOverflowWraps(t *testing.T) {
	n := int64(math.MaxInt64)
	n++
	if n != math.MinInt64 {
		t.Fatalf("got %d", n)
	}
}

/// Three ways to add one to the largest integer: a debug build panics on plain `+`,
/// a release build wraps, and `checked_add` returns `None` in both. kv uses the last.
#[test]
fn overflow_three_ways() {
    let n = i64::MAX;
    assert_eq!(n.checked_add(1), None);
    assert_eq!(n.wrapping_add(1), i64::MIN);
    assert_eq!(n.saturating_add(1), i64::MAX);
}

#[test]
#[cfg(debug_assertions)]
#[should_panic(expected = "attempt to add with overflow")]
fn plain_add_panics_in_debug() {
    let n = std::hint::black_box(i64::MAX);
    let _ = n + 1;
}

/* FNV-1a: a small, fast hash of bytes. Not safe against an attacker who chooses keys
 * to collide, which Redis guards against with a secret seed. Chapter 12 returns to it. */
static uint64_t kv_hash(const char *s, size_t n) {
    uint64_t h = 14695981039346656037ULL;
    for (size_t i = 0; i < n; i++) {
        h ^= (unsigned char)s[i];
        h *= 1099511628211ULL;
    }
    return h;
}

/* One key and its value, in a chain of entries that share a bucket. The key and the
 * value are owned copies. The next pointer is owned too: freeing an entry frees its
 * chain. */
typedef struct kv_entry {
    char *key;
    size_t klen;
    char *val;
    size_t vlen;
    struct kv_entry *next;
} kv_entry;

/* A hash table with separate chaining. buckets[i] is the head of a chain, or NULL. The
 * table grows when it holds more entries than buckets, so a chain stays short. */
typedef struct {
    kv_entry **buckets; /* owned array of cap pointers */
    size_t cap;         /* a power of two, or 0 before the first insert */
    size_t len;         /* how many entries, over all chains */
} kv_store;

static kv_entry *kv_store_find(kv_store *s, const char *key, size_t klen) {
    if (s->cap == 0) {
        return NULL;
    }
    size_t i = kv_hash(key, klen) & (s->cap - 1);
    for (kv_entry *e = s->buckets[i]; e != NULL; e = e->next) {
        if (kv_bytes_eq(e->key, e->klen, key, klen)) {
            return e;
        }
    }
    return NULL;
}
/* The command list, written once. Each use of KV_COMMANDS defines X to say what to make
 * of a row. The first use builds the table, the second counts it. A positive arity is
 * the exact argc, name included, and a negative one is a minimum, as Redis counts. */
#define KV_COMMANDS                     \
    X(ping, kv_cmd_ping, -1)            \
    X(echo, kv_cmd_echo, 2)             \
    X(set, kv_cmd_set, 3)               \
    X(get, kv_cmd_get, 2)               \
    X(del, kv_cmd_del, -2)              \
    X(exists, kv_cmd_exists, -2)        \
    X(incr, kv_cmd_incr, 2)             \
    X(append, kv_cmd_append, 3)         \
    X(strlen, kv_cmd_strlen, 2)         \
    X(getrange, kv_cmd_getrange, 4)     \
    X(dbsize, kv_cmd_dbsize, 1)         \
    X(keys, kv_cmd_keys, 2)             \
    X(command, kv_cmd_command, -1)

typedef struct {
    const char *name;
    kv_cmd_fn fn;
    int arity;
} kv_command;

static const kv_command kv_commands[] = {
#define X(name, fn, arity) {#name, fn, arity},
    KV_COMMANDS
#undef X
};

enum {
#define X(name, fn, arity) +1
    KV_COMMAND_COUNT = 0 KV_COMMANDS
#undef X
};

/* Double the bucket array and move every entry to its new chain. No entry is copied or
 * freed: only the next pointers change. */
static bool kv_store_grow(kv_store *s) {
    size_t cap = s->cap ? s->cap * 2 : 16;
    kv_entry **buckets = calloc(cap, sizeof *buckets);
    if (buckets == NULL) {
        return false;
    }
    for (size_t i = 0; i < s->cap; i++) {
        kv_entry *e = s->buckets[i];
        while (e != NULL) {
            kv_entry *next = e->next; /* read it before the chain is changed */
            size_t j = kv_hash(e->key, e->klen) & (cap - 1);
            e->next = buckets[j];
            buckets[j] = e;
            e = next;
        }
    }
    free(s->buckets);
    s->buckets = buckets;
    s->cap = cap;
    return true;
}

/* Store a copy of val under a copy of key, replacing any old value. */
static bool kv_store_set(kv_store *s, const char *key, size_t klen, const char *val,
                         size_t vlen) {
    char *v = kv_bytes_dup(val, vlen);
    if (v == NULL) {
        return false;
    }
    kv_entry *e = kv_store_find(s, key, klen);
    if (e != NULL) {
        free(e->val);
        e->val = v;
        e->vlen = vlen;
        return true;
    }
    if (s->len >= s->cap && !kv_store_grow(s)) {
        free(v);
        return false;
    }
    e = malloc(sizeof *e);
    char *k = kv_bytes_dup(key, klen);
    if (e == NULL || k == NULL) {
        free(e);
        free(k);
        free(v);
        return false;
    }
    size_t i = kv_hash(key, klen) & (s->cap - 1);
    *e = (kv_entry){k, klen, v, vlen, s->buckets[i]};
    s->buckets[i] = e;
    s->len++;
    return true;
}

/* KEYS pattern: every matching key, as an array. The count is known only after the
 * walk, so the keys are collected first and the header is written in front of them. */
static bool kv_cmd_keys(const kv_cmd *c, kv_store *s, kv_buf *out) {
    kv_buf keys = {0};
    size_t n = 0;
    for (size_t i = 0; i < s->cap; i++) {
        for (kv_entry *e = s->buckets[i]; e != NULL; e = e->next) {
            if (kv_glob(c->argv[1], c->arglen[1], e->key, e->klen)) {
                if (!kv_reply_bulk(&keys, e->key, e->klen)) {
                    kv_buf_free(&keys);
                    return false;
                }
                n++;
            }
        }
    }
    bool ok = kv_reply_array(out, n) && kv_buf_append(out, keys.data, keys.len);
    kv_buf_free(&keys);
    return ok;
}

/* 1. A void pointer and an element size. The caller says how big one element is, and the
 * function moves bytes without knowing what they are. qsort and bsearch work this way.
 * The compiler checks nothing: pass the wrong size and the sort is silently wrong. */
static int cmp_u32(const void *a, const void *b) {
    uint32_t x = *(const uint32_t *)a;
    uint32_t y = *(const uint32_t *)b;
    return (x > y) - (x < y);
}

static void sort_demo(void) {
    uint32_t xs[] = {30, 10, 20};
    qsort(xs, 3, sizeof xs[0], cmp_u32);
    assert(xs[0] == 10 && xs[1] == 20 && xs[2] == 30);
}

/* 2. A macro that writes a typed container for each element type. The macro is the
 * template, and each use is a copy with T filled in. A Rust generic compiles the same
 * way, one copy per type, with the difference that Rust checks the template once. */
#define DEFINE_VEC(name, T)                                                           \
    typedef struct {                                                                  \
        T *data; /* owned */                                                          \
        size_t len, cap;                                                              \
    } name;                                                                           \
    static bool name##_push(name *v, T x) {                                           \
        if (v->len == v->cap) {                                                       \
            size_t cap = v->cap ? v->cap * 2 : 8;                                     \
            T *p = realloc(v->data, cap * sizeof *p);                                 \
            if (p == NULL) {                                                          \
                return false;                                                         \
            }                                                                         \
            v->data = p;                                                              \
            v->cap = cap;                                                             \
        }                                                                             \
        v->data[v->len++] = x;                                                        \
        return true;                                                                  \
    }

DEFINE_VEC(vec_u32, uint32_t)
DEFINE_VEC(vec_str, const char *)

static void vec_demo(void) {
    vec_u32 nums = {0};
    vec_str words = {0};
    for (uint32_t i = 0; i < 100; i++) {
        assert(vec_u32_push(&nums, i * i));
    }
    assert(vec_str_push(&words, "hello") && vec_str_push(&words, "world"));
    assert(nums.len == 100 && nums.data[9] == 81);
    assert(strcmp(words.data[1], "world") == 0);
    free(nums.data);
    free(words.data);
}

/* 3. _Generic: a compile-time switch on the type of an expression. It is how C11 gives
 * one name to several functions, which is as close as C gets to overloading. */
static const char *name_u32(uint32_t x) {
    (void)x;
    return "uint32_t";
}

static const char *name_str(const char *s) {
    (void)s;
    return "const char *";
}

#define type_name(x) _Generic((x), uint32_t: name_u32, const char *: name_str, char *: name_str)(x)

static void generic_demo(void) {
    uint32_t n = 7;
    const char *s = "seven";
    assert(strcmp(type_name(n), "uint32_t") == 0);
    assert(strcmp(type_name(s), "const char *") == 0);
}

/* What a struct is made of. Every type has an alignment, and the compiler inserts padding
 * so that each member starts at a multiple of its own alignment. One byte of tag before
 * an eight-byte number costs seven bytes of padding, and reordering the members saves it.
 * offsetof reads the layout, and _Static_assert checks it when the program is compiled. */
typedef struct {
    uint8_t tag;
    uint64_t n;
    uint8_t flag;
} loose;

typedef struct {
    uint64_t n;
    uint8_t tag;
    uint8_t flag;
} tight;

_Static_assert(sizeof(loose) == 24, "two bytes of tags cost 14 bytes of padding");
_Static_assert(sizeof(tight) == 16, "the same members, reordered, need 6");
_Static_assert(offsetof(loose, n) == 8, "n starts at the next multiple of 8");
_Static_assert(alignof(uint64_t) == 8, "a 64-bit integer is aligned to 8 bytes");

static void layout_demo(void) {
    printf("loose %zu bytes, tight %zu bytes, char is %s\n", sizeof(loose), sizeof(tight),
           CHAR_MIN < 0 ? "signed" : "unsigned");
}

/* An arena: one block, handed out from the front, freed all at once. Each piece starts
 * at a multiple of the largest alignment, so any type may live in it. A parser that
 * builds a tree per request, or a game that frees everything per frame, uses one. */
typedef struct {
    char *base; /* owned */
    size_t used;
    size_t cap;
} arena;

static void *arena_alloc(arena *a, size_t n) {
    size_t align = alignof(max_align_t);
    size_t start = (a->used + align - 1) / align * align; /* round up */
    if (start + n > a->cap) {
        return NULL;
    }
    a->used = start + n;
    return a->base + start;
}

static void arena_demo(void) {
    arena a = {malloc(4096), 0, 4096};
    assert(a.base != NULL);
    char *name = arena_alloc(&a, 5);
    uint64_t *count = arena_alloc(&a, sizeof *count);
    memcpy(name, "kv!!", 5);
    *count = 42;
    assert((uintptr_t)count % alignof(uint64_t) == 0); /* padded past the 5 bytes */
    assert(a.used == 16 + sizeof *count);
    free(a.base); /* one free for every piece */
}

// fnv1a hashes bytes the way kv does in C, so the two tables spread keys the same way.
func fnv1a(b []byte) uint64 {
	h := uint64(14695981039346656037)
	for _, c := range b {
		h ^= uint64(c)
		h *= 1099511628211
	}
	return h
}

// entry is one key and its value in a chain. There is no next pointer: a chain is a
// slice, and the garbage collector owns everything. K and V are type parameters: the
// table is written once, for any key that supports == and any value at all.
type entry[K comparable, V any] struct {
	key K
	val V
}

// Table is a hash table with separate chaining. It grows when it holds more entries
// than buckets, so a chain stays short. The hash function comes from the caller, because
// the constraint comparable promises == and nothing else.
type Table[K comparable, V any] struct {
	hash    func(K) uint64
	buckets [][]entry[K, V]
	n       int
}

func NewTable[K comparable, V any](hash func(K) uint64) *Table[K, V] {
	return &Table[K, V]{hash: hash}
}

func (t *Table[K, V]) Get(key K) (V, bool) {
	if len(t.buckets) == 0 {
		var zero V
		return zero, false
	}
	chain := t.buckets[t.hash(key)&uint64(len(t.buckets)-1)]
	for i := range chain {
		if chain[i].key == key {
			return chain[i].val, true
		}
	}
	var zero V
	return zero, false
}

func (t *Table[K, V]) Del(key K) bool {
	if len(t.buckets) == 0 {
		return false
	}
	i := t.hash(key) & uint64(len(t.buckets)-1)
	chain := t.buckets[i]
	for j := range chain {
		if chain[j].key == key {
			// move the last entry into the hole, then shorten the chain by one
			chain[j] = chain[len(chain)-1]
			t.buckets[i] = chain[:len(chain)-1]
			t.n--
			return true
		}
	}
	return false
}

// Len is the number of entries, and Keys is every key in no particular order.
func (t *Table[K, V]) Len() int { return t.n }

func (t *Table[K, V]) Keys() []K {
	keys := make([]K, 0, t.n)
	for _, chain := range t.buckets {
		for _, e := range chain {
			keys = append(keys, e.key)
		}
	}
	return keys
}

// TestTableAgainstMap runs random operations on the table and on a map, and checks
// after each one that the table agrees with the map. The table is instantiated with
// string keys and string values; the compiler writes that version of the code.
func TestTableAgainstMap(t *testing.T) {
	tb := NewTable[string, string](hashString)
	ref := map[string]string{}
	r := rand.New(rand.NewPCG(1, 2))
	for step := range 5000 {
		key := fmt.Sprintf("k%d", r.IntN(200))
		switch r.IntN(3) {
		case 0:
			val := fmt.Sprintf("v%d", step)
			tb.Set(key, val)
			ref[key] = val
		case 1:
			_, want := ref[key]
			if got := tb.Del(key); got != want {
				t.Fatalf("step %d: del %q = %v, want %v", step, key, got, want)
			}
			delete(ref, key)
		case 2:
			want, ok := ref[key]
			got, found := tb.Get(key)
			if found != ok || got != want {
				t.Fatalf("step %d: get %q = %q,%v, want %q,%v", step, key, got, found, want, ok)
			}
		}
		if tb.Len() != len(ref) {
			t.Fatalf("step %d: table has %d entries, map has %d", step, tb.Len(), len(ref))
		}
	}
	// the generic slices and maps packages: collect the map's keys, sort both, compare
	want := slices.Sorted(maps.Keys(ref))
	got := tb.Keys()
	slices.Sort(got)
	if !slices.Equal(got, want) {
		t.Fatalf("keys differ: %v vs %v", got, want)
	}
}

// TestTableWithIntKeys is the same table with another key type. Nothing in table.go
// changed; the compiler made a second copy for int keys.
func TestTableWithIntKeys(t *testing.T) {
	tb := NewTable[int, []byte](func(k int) uint64 { return uint64(k) * 11400714819323198485 })
	for i := range 100 {
		tb.Set(i, []byte{byte(i)})
	}
	if v, ok := tb.Get(42); !ok || v[0] != 42 {
		t.Fatalf("get 42 = %v, %v", v, ok)
	}
	if tb.Del(42) != true || tb.Len() != 99 {
		t.Fatal("del")
	}
}

// store keeps every key and its value in a map. The hand-written table in table.go
// would work too, and the tests show it does, but the map is faster and nobody would
// use anything else in Go.
type store struct {
	m map[string][]byte
}

func newStore() *store {
	return &store{m: make(map[string][]byte)}
}

func (s *store) get(key []byte) ([]byte, bool) {
	v, ok := s.m[string(key)]
	return v, ok
}

func (s *store) set(key, val []byte) {
	s.m[string(key)] = val
}

func (s *store) del(key []byte) bool {
	_, ok := s.m[string(key)]
	delete(s.m, string(key))
	return ok
}

func (s *store) size() int {
	return len(s.m)
}

// keys returns every key that matches the pattern, in no particular order.
func (s *store) keys(pattern []byte) [][]byte {
	var out [][]byte
	for k := range s.m {
		if glob(pattern, []byte(k)) {
			out = append(out, []byte(k))
		}
	}
	return out
}

/// FNV-1a, the same hash the C version of kv uses.
fn fnv1a(bytes: &[u8]) -> u64 {
    bytes.iter().fold(0xcbf2_9ce4_8422_2325, |h, &b| {
        (h ^ u64::from(b)).wrapping_mul(0x0100_0000_01b3)
    })
}

/// One key and its value in a chain. The chain is a Vec, which owns its entries, and
/// the entries own their bytes. Dropping the table drops all of it. `K` and `V` are type
/// parameters: the table is written once for any key that can be seen as bytes and
/// compared, and any value at all.
struct Entry<K, V> {
    key: K,
    val: V,
}

/// A hash table with separate chaining. It grows when it holds more entries than
/// buckets, so a chain stays short.
pub struct Table<K, V> {
    buckets: Vec<Vec<Entry<K, V>>>,
    len: usize,
}

/// An empty table, for every `K` and `V`. `derive(Default)` would also demand that `K`
/// and `V` have defaults, which an empty table does not need, so this is written out.
impl<K, V> Default for Table<K, V> {
    fn default() -> Self {
        Self {
            buckets: Vec::new(),
            len: 0,
        }
    }
}

/// The bounds say what the table needs from a key: a view of it as bytes, to hash, and
/// equality, to find it. A key type with more is fine; one with less does not compile.
impl<K, V> Table<K, V>
where
    K: AsRef<[u8]> + PartialEq,
{
    fn index(&self, key: &K) -> usize {
        (fnv1a(key.as_ref()) & (self.buckets.len() as u64 - 1)) as usize
    }

    pub fn get(&self, key: &K) -> Option<&V> {
        if self.buckets.is_empty() {
            return None;
        }
        let chain = &self.buckets[self.index(key)];
        chain.iter().find(|e| e.key == *key).map(|e| &e.val)
    }

    pub fn del(&mut self, key: &K) -> bool {
        if self.buckets.is_empty() {
            return false;
        }
        let i = self.index(key);
        let chain = &mut self.buckets[i];
        match chain.iter().position(|e| e.key == *key) {
            Some(j) => {
                // move the last entry into the hole; the removed entry is dropped here
                chain.swap_remove(j);
                self.len -= 1;
                true
            }
            None => false,
        }
    }

    pub fn len(&self) -> usize {
        self.len
    }

    pub fn is_empty(&self) -> bool {
        self.len == 0
    }
}

#[cfg(test)]
mod tests {
    use super::Table;
    use std::collections::HashMap;

    /// A small deterministic random number generator, so the test is the same every run.
    fn next(state: &mut u64) -> u64 {
        *state ^= *state << 13;
        *state ^= *state >> 7;
        *state ^= *state << 17;
        *state
    }

    #[test]
    fn agrees_with_hashmap() {
        let mut table: Table<Vec<u8>, Vec<u8>> = Table::default();
        let mut reference: HashMap<Vec<u8>, Vec<u8>> = HashMap::new();
        let mut rng = 0x2545_f491_4f6c_dd1d_u64;
        for step in 0..5000 {
            let key = format!("k{}", next(&mut rng) % 200).into_bytes();
            match next(&mut rng) % 3 {
                0 => {
                    let val = format!("v{step}").into_bytes();
                    table.set(key.clone(), val.clone());
                    reference.insert(key, val);
                }
                1 => {
                    assert_eq!(
                        table.del(&key),
                        reference.remove(&key).is_some(),
                        "step {step}"
                    );
                }
                _ => {
                    assert_eq!(table.get(&key), reference.get(&key));
                }
            }
            assert_eq!(table.len(), reference.len());
        }
    }

    /// The same table with another key type. `&str` is bytes with `as_ref`, and a
    /// `u32` value needs nothing. The compiler made a second copy of every method.
    #[test]
    fn works_with_str_keys_and_integer_values() {
        let mut table: Table<&str, u32> = Table::default();
        for (i, word) in ["one", "two", "three"].iter().enumerate() {
            table.set(word, i as u32);
        }
        assert_eq!(table.get(&"two"), Some(&1));
        assert!(table.del(&"two"));
        assert_eq!(table.len(), 2);
    }

/// Two owners of one table, in one thread. `Rc` counts the owners and frees the table
/// when the last one goes; `RefCell` moves the borrow rules to run time, so each
/// owner can borrow it mutably in turn. A second mutable borrow at the same moment
/// is a panic, not a compile error: the check moved, it did not go away.
#[test]
fn two_owners_with_rc_and_refcell() {
    use std::cell::{Cell, RefCell};
    use std::rc::Rc;

    let shared: Rc<RefCell<Table<&str, u32>>> = Rc::new(RefCell::new(Table::default()));
    let writer = Rc::clone(&shared);
    let reader = Rc::clone(&shared);
    assert_eq!(Rc::strong_count(&shared), 3);

    writer.borrow_mut().set("hits", 1);
    assert_eq!(reader.borrow().get(&"hits"), Some(&1));
    drop(writer);
    assert_eq!(Rc::strong_count(&shared), 2);

    // Cell is the same idea for a Copy value: change it through a shared reference
    let counter = Cell::new(0);
    let bump = |c: &Cell<u32>| c.set(c.get() + 1);
    bump(&counter);
    bump(&counter);
    assert_eq!(counter.get(), 2);

    // the run-time check: a borrow_mut while a borrow is live
    let guard = shared.borrow();
    assert!(shared.try_borrow_mut().is_err());
    drop(guard);
    assert!(shared.try_borrow_mut().is_ok());
}

#ifndef KV_BUF_H
#define KV_BUF_H

#include <stdbool.h>
#include <stddef.h>

/* A growable byte buffer. It owns data, and kv_buf_free releases it. Start one with
 * kv_buf b = {0}. */
typedef struct {
    char *data; /* owned; NULL until the first append */
    size_t len;
    size_t cap;
} kv_buf;

bool kv_buf_append(kv_buf *b, const void *src, size_t n);
void kv_buf_consume(kv_buf *b, size_t n);
void kv_buf_free(kv_buf *b);

#endif

/* Private: only this file calls it, so it is static and not in the header. */
static bool kv_buf_reserve(kv_buf *b, size_t extra) {
    if (b->len + extra <= b->cap) {
        return true;
    }
    size_t cap = b->cap ? b->cap * 2 : 1024;
    while (cap < b->len + extra) {
        cap *= 2;
    }
    char *p = realloc(b->data, cap);
    if (p == NULL) {
        return false; /* the old block is still valid and still ours */
    }
    b->data = p;
    b->cap = cap;
    return true;
}

/* The store itself is opaque: this header says it exists, store.c says what is in it.
 * Other files hold a pointer and call the functions below, and cannot reach inside. */
typedef struct kv_store kv_store;

/* A new, empty store of the given type, or NULL when out of memory. Free it with
 * kv_store_free, which also frees every entry. */
kv_store *kv_store_new(const kv_store_type *type);
void kv_store_free(kv_store *s);

size_t kv_store_len(const kv_store *s);

/* The entry for key, or NULL. The pointer is valid until the next set or del. */
kv_entry *kv_store_find(kv_store *s, const char *key, size_t klen);

/* Store a copy of val under a copy of key, replacing any old value. */
bool kv_store_set(kv_store *s, const char *key, size_t klen, const char *val, size_t vlen);
bool kv_store_del(kv_store *s, const char *key, size_t klen);

/* Call fn once for every entry, with ctx passed through, until fn returns false. The
 * callback must not set or del: the chain it is walking would move under it. */
void kv_store_each(kv_store *s, bool (*fn)(kv_entry *e, void *ctx), void *ctx);

void kv_store_each(kv_store *s, bool (*fn)(kv_entry *e, void *ctx), void *ctx) {
    for (size_t i = 0; i < s->cap; i++) {
        for (kv_entry *e = s->buckets[i]; e != NULL; e = e->next) {
            if (!fn(e, ctx)) {
                return;
            }
        }
    }
}

/* What the KEYS walk needs to know, passed to the callback through a void pointer. */
typedef struct {
    const char *pat;
    size_t plen;
    kv_buf keys; /* the matching keys, as replies, collected before the count is known */
    size_t n;
    bool ok;
} kv_keys_ctx;

/* Called once per entry by kv_store_each. The context comes back as a void pointer, and
 * the first line turns it into the real type again. */
static bool kv_keys_visit(kv_entry *e, void *ctx) {
    kv_keys_ctx *k = ctx;
    if (!kv_glob(k->pat, k->plen, e->key, e->klen)) {
        return true;
    }
    if (!kv_reply_bulk(&k->keys, e->key, e->klen)) {
        k->ok = false;
        return false; /* stop the walk */
    }
    k->n++;
    return true;
}

/* KEYS pattern: every matching key, as an array. The walk is the store's; this file only
 * supplies what to do with each entry. */
static bool kv_cmd_keys(const kv_cmd *c, kv_store *s, kv_buf *out) {
    kv_keys_ctx k = {c->argv[1], c->arglen[1], {0}, 0, true};
    kv_store_each(s, kv_keys_visit, &k);
    bool ok = k.ok && kv_reply_array(out, k.n) && kv_buf_append(out, k.keys.data, k.keys.len);
    kv_buf_free(&k.keys);
    return ok;
}

/* How the table treats a key: a struct of function pointers, which is how C describes
 * behaviour a caller may choose. A store is made with one of the types below, and every
 * hash and comparison inside it goes through that type. Redis calls the same thing a
 * dictType. */
typedef struct {
    const char *name;
    uint64_t (*hash)(const char *key, size_t klen, uint64_t seed);
    bool (*eq)(const char *a, size_t alen, const char *b, size_t blen);
    bool seeded; /* kv_store_new draws a random seed, so keys cannot be chosen to collide */
} kv_store_type;

extern const kv_store_type kv_fnv_type;    /* FNV-1a, the same hash on every run */
extern const kv_store_type kv_seeded_type; /* FNV-1a mixed with a seed drawn at start */

/* FNV-1a: a small, fast hash of bytes. The seed is mixed into the starting value, so a
 * seeded store hashes the same key differently on every run. */
static uint64_t kv_fnv1a(const char *s, size_t n, uint64_t seed) {
    uint64_t h = 14695981039346656037ULL ^ seed;
    for (size_t i = 0; i < n; i++) {
        h ^= (unsigned char)s[i];
        h *= 1099511628211ULL;
    }
    return h;
}

const kv_store_type kv_fnv_type = {"fnv", kv_fnv1a, kv_bytes_eq, false};
const kv_store_type kv_seeded_type = {"seeded", kv_fnv1a, kv_bytes_eq, true};

/* Eight random bytes from the operating system, or the clock when that fails. */
static uint64_t kv_random_seed(void) {
    uint64_t seed = 0;
    FILE *f = fopen("/dev/urandom", "rb");
    if (f != NULL) {
        size_t got = fread(&seed, sizeof seed, 1, f);
        fclose(f);
        if (got == 1) {
            return seed;
        }
    }
    return (uint64_t)(uintptr_t)&seed * 2654435761ULL;
}

kv_store *kv_store_new(const kv_store_type *type) {
    kv_store *s = calloc(1, sizeof *s);
    if (s == NULL) {
        return NULL;
    }
    s->type = type;
    s->seed = type->seeded ? kv_random_seed() : 0;
    return s;
}

size_t kv_store_len(const kv_store *s) {
    return s->len;
}

#include "buf.h"
#include "cmd.h"
#include "resp.h"
#include "store.h"

/* KV_HASH=fnv asks for the unseeded hash, which makes a run repeatable. The default
 * seeds the hash, so that nobody can choose keys that all land in one chain. */
const char *hash = getenv("KV_HASH");
const kv_store_type *type = hash && strcmp(hash, "fnv") == 0 ? &kv_fnv_type : &kv_seeded_type;
kv_store *store = kv_store_new(type);
if (store == NULL) {
    perror("kv_store_new");
    close(fd);
    return 1;
}

CC       ?= cc
CFLAGS   ?= -std=c17 -Wall -Wextra -Werror -O2 -g
SANITIZE ?=

# every .c under src/ becomes a .o under build/, and kv is linked from all of them
SRC := $(wildcard src/*.c)
OBJ := $(SRC:src/%.c=build/%.o)

kv: $(OBJ)
	$(CC) $(CFLAGS) $(SANITIZE) -o $@ $(OBJ)

# -MMD writes a .d file beside each .o that lists the headers it included, so that a
# change to a header rebuilds exactly the files that use it
build/%.o: src/%.c | build
	$(CC) $(CFLAGS) $(SANITIZE) -MMD -c -o $@ $<

build:
	mkdir -p build

-include $(OBJ:.o=.d)

clean:
	rm -rf kv kv.dSYM build

.PHONY: clean

// ReadCommand reads one request: an array of bulk strings, or one inline line of words.
// It returns io.EOF when the client closed the connection between commands.
func ReadCommand(r *bufio.Reader) ([][]byte, error) {
	line, err := readLine(r)
	if err != nil {
		return nil, fmt.Errorf("read command: %w", err)
	}
	if len(line) == 0 || line[0] != '*' {
		return bytes.Fields(line), nil
	}
	n, err := strconv.Atoi(string(line[1:]))
	if err != nil || n < 0 || n > maxArgs {
		return nil, protoErr('*')
	}
	args := make([][]byte, 0, n)
	for range n {
		line, err := readLine(r)
		if err != nil {
			return nil, fmt.Errorf("read argument: %w", unexpected(err))
		}
		if len(line) == 0 || line[0] != '$' {
			return nil, protoErr(first(line))
		}
		size, err := strconv.Atoi(string(line[1:]))
		if err != nil || size < 0 || size > maxBulk {
			return nil, protoErr('$')
		}
		buf := make([]byte, size+2)
		if _, err := io.ReadFull(r, buf); err != nil {
			return nil, fmt.Errorf("read argument: %w", unexpected(err))
		}
		if buf[size] != '\r' || buf[size+1] != '\n' {
			return nil, protoErr(buf[size])
		}
		args = append(args, buf[:size])
	}
	return args, nil
}

// Execute runs one command against the store and writes its reply. The two imports
// above are this module's own packages: kv is the module path from go.mod.
func Execute(args [][]byte, s Storer, w *bufio.Writer) error {

// Store keeps every key and its value in a map. Exported, so that package cmd can use
// it; its field is not, so that only this package can touch the map.
type Store struct {
	m map[string][]byte
}

func New() *Store {
	return &Store{m: make(map[string][]byte)}
}

func (s *Store) Get(key []byte) ([]byte, bool) {
	v, ok := s.m[string(key)]
	return v, ok
}

func (s *Store) Set(key, val []byte) {
	s.m[string(key)] = val
}

func (s *Store) Del(key []byte) bool {
	_, ok := s.m[string(key)]
	delete(s.m, string(key))
	return ok
}

func (s *Store) Len() int {
	return len(s.m)
}

// Keys returns every key that matches the pattern, in no particular order.
func (s *Store) Keys(pattern []byte) [][]byte {
	var out [][]byte
	for k := range s.m {
		if glob(pattern, []byte(k)) {
			out = append(out, []byte(k))
		}
	}
	return out
}

// A Storer is whatever the commands need from a store: these six methods, and nothing
// about how they are done. Any type with these methods is a Storer, without saying so.
// The interface lives here, with the code that uses it, which is where Go puts one.
type Storer interface {
	Get(key []byte) ([]byte, bool)
	Set(key, val []byte)
	Del(key []byte) bool
	Len() int
	Keys(pattern []byte) [][]byte
	Incr(key []byte) (int64, error)
}

// ByteTable is the hand-written table from chapter 6, used as a store. It has the same
// methods as Store, so it satisfies the cmd.Storer interface without naming it. Nothing
// here says "implements": the methods are the whole of the contract.
type ByteTable struct {
	t *Table[string, []byte]
}

func NewByteTable() *ByteTable {
	return &ByteTable{t: NewTable[string, []byte](func(s string) uint64 { return fnv1a([]byte(s)) })}
}

func (b *ByteTable) Get(key []byte) ([]byte, bool) { return b.t.Get(string(key)) }
func (b *ByteTable) Set(key, val []byte)           { b.t.Set(string(key), val) }
func (b *ByteTable) Del(key []byte) bool           { return b.t.Del(string(key)) }
func (b *ByteTable) Len() int                      { return b.t.Len() }

func (b *ByteTable) Keys(pattern []byte) [][]byte {
	var out [][]byte
	for _, k := range b.t.Keys() {
		if glob(pattern, []byte(k)) {
			out = append(out, []byte(k))
		}
	}
	return out
}

func (b *ByteTable) Incr(key []byte) (int64, error) {
	var n int64
	if cur, ok := b.Get(key); ok {
		var err error
		n, err = strconv.ParseInt(string(cur), 10, 64)
		if err != nil {
			return 0, ErrNotInteger
		}
	}
	if n == math.MaxInt64 {
		return 0, ErrOverflow
	}
	n++
	b.Set(key, []byte(strconv.FormatInt(n, 10)))
	return n, nil
}

// Either store satisfies cmd.Storer. The variable's type is the interface, and the
// value inside it is whichever the environment asked for.
var s cmd.Storer = store.New()
if os.Getenv("KV_STORE") == "table" {
	s = store.NewByteTable()
}

// counting wraps any Storer and counts the writes. The embedded field lends the wrapped
// store's methods to counting, so only Set is written here: the rest are promoted.
type counting struct {
	Storer
	sets int
}

func (c *counting) Set(key, val []byte) {
	c.sets++
	c.Storer.Set(key, val)
}

func TestEmbeddedStorerPromotesMethods(t *testing.T) {
	c := &counting{Storer: store.New()}
	var buf bytes.Buffer
	w := bufio.NewWriter(&buf)
	for _, line := range [][]string{{"SET", "a", "1"}, {"SET", "b", "2"}, {"GET", "a"}} {
		var args [][]byte
		for _, s := range line {
			args = append(args, []byte(s))
		}
		if err := Execute(args, c, w); err != nil {
			t.Fatal(err)
		}
	}
	w.Flush()
	if c.sets != 2 || c.Len() != 2 || !bytes.Contains(buf.Bytes(), []byte("$1\r\n1\r\n")) {
		t.Fatalf("sets %d, len %d, out %q", c.sets, c.Len(), buf.String())
	}
}

// A nil pointer inside an interface is not a nil interface. The interface holds a type
// and a pointer; the type is set, so the interface is not nil, and a method call reaches
// the pointer receiver with nil. This is the trap every Go programmer meets once.
func TestNilPointerInInterfaceIsNotNil(t *testing.T) {
	var p *store.Store // nil
	var s Storer = p   // a Storer holding (*store.Store, nil)
	if s == nil {
		t.Fatal("an interface holding a typed nil compares unequal to nil")
	}
	var none Storer // no type, no value
	if none != nil {
		t.Fatal("an interface with no type is nil")
	}
}

pub mod cmd;
pub mod resp;
pub mod store;
pub mod table;

use kv::store::{MapStore, Store};
use kv::table::Table;
use kv::{cmd, resp};

/// What the commands need from a store. A trait is a set of methods a type promises, and
/// any type that writes an `impl Store for` block with these methods is a store. The
/// server is written against the trait, so the type behind it can change.
pub trait Store {
    /// A borrowed view of the value, which lives as long as the borrow of the store.
    fn get(&self, key: &[u8]) -> Option<&[u8]>;
    fn set(&mut self, key: Vec<u8>, val: Vec<u8>);
    fn del(&mut self, key: &[u8]) -> bool;
    fn len(&self) -> usize;
    /// Every key that matches the pattern, in no particular order.
    fn keys(&self, pattern: &[u8]) -> Vec<&[u8]>;

    fn is_empty(&self) -> bool {
        self.len() == 0
    }

    /// Add one to the integer stored under key, which counts as 0 when absent. A default
    /// method: written once here, in terms of `get` and `set`, and shared by every
    /// implementation unless one replaces it.
    fn incr(&mut self, key: &[u8]) -> Result<i64, IncrError> {
        let n = match self.get(key) {
            None => 0,
            Some(bytes) => parse_int(bytes).ok_or(IncrError::NotInteger)?,
        };
        let n = n.checked_add(1).ok_or(IncrError::Overflow)?;
        self.set(key.to_vec(), n.to_string().into_bytes());
        Ok(n)
    }
}

/// The store the server uses: the standard HashMap. The hand-written table in table.rs
/// implements the same trait, and `KV_STORE=table` runs the server on it.
#[derive(Default)]
pub struct MapStore {
    map: HashMap<Vec<u8>, Vec<u8>>,
}

impl MapStore {
    pub fn new() -> Self {
        Self::default()
    }
}

impl Store for MapStore {
    fn get(&self, key: &[u8]) -> Option<&[u8]> {
        self.map.get(key).map(Vec::as_slice)
    }

    fn set(&mut self, key: Vec<u8>, val: Vec<u8>) {
        self.map.insert(key, val);
    }

    fn del(&mut self, key: &[u8]) -> bool {
        self.map.remove(key).is_some()
    }

    fn len(&self) -> usize {
        self.map.len()
    }

    fn keys(&self, pattern: &[u8]) -> Vec<&[u8]> {
        self.map
            .keys()
            .map(Vec::as_slice)
            .filter(|k| glob(pattern, k))
            .collect()
    }
}

/// The hand-written table as a store: the same trait as MapStore, so the server can run
/// on either. The methods are those of the trait; the body of each is one line.
impl crate::store::Store for Table<Vec<u8>, Vec<u8>> {
    fn get(&self, key: &[u8]) -> Option<&[u8]> {
        Table::get(self, &key.to_vec()).map(Vec::as_slice)
    }

    fn set(&mut self, key: Vec<u8>, val: Vec<u8>) {
        Table::set(self, key, val);
    }

    fn del(&mut self, key: &[u8]) -> bool {
        Table::del(self, &key.to_vec())
    }

    fn len(&self) -> usize {
        Table::len(self)
    }

    fn keys(&self, pattern: &[u8]) -> Vec<&[u8]> {
        self.buckets
            .iter()
            .flatten()
            .map(|e| e.key.as_slice())
            .filter(|k| crate::store::glob(pattern, k))
            .collect()
    }
}

// Either type implements Store. A Box<dyn Store> holds whichever one the environment
// asked for, and every call through it is decided at run time.
let mut store: Box<dyn Store> = match std::env::var("KV_STORE").as_deref() {
    Ok("table") => Box::new(Table::default()),
    _ => Box::new(MapStore::new()),
};

#[cfg(test)]
mod tests {
    use super::{MapStore, Store};
    use crate::table::Table;

    /// One test for every store. `S: Store` is a bound, and the compiler writes a copy of
    /// this function for each type it is called with: static dispatch, no trait object.
    fn behaves_like_a_store<S: Store>(mut store: S) {
        assert!(store.is_empty());
        store.set(b"a".to_vec(), b"1".to_vec());
        assert_eq!(store.get(b"a"), Some(&b"1"[..]));
        assert_eq!(store.incr(b"a"), Ok(2));
        assert_eq!(store.keys(b"*").len(), 1);
        assert!(store.del(b"a"));
        assert!(store.get(b"a").is_none());
    }

    #[test]
    fn the_map_store_is_a_store() {
        behaves_like_a_store(MapStore::new());
    }

    #[test]
    fn the_table_is_a_store() {
        behaves_like_a_store(Table::<Vec<u8>, Vec<u8>>::default());
    }
}

# The layout every C file in kv keeps. clang-format -i src/*.c src/*.h applies it.
BasedOnStyle: LLVM
IndentWidth: 4
ColumnLimit: 100
AllowShortFunctionsOnASingleLine: Inline

/* Which of the four shapes a value has. The number is the tag of the union below. */
typedef enum { KV_STRING, KV_LIST, KV_HASH, KV_SET } kv_kind;

/* A tagged union: the tag says which member of the union is live, and reading another
 * member is a bug the compiler will not catch. A hash and a set are each a nested store,
 * whose entries hold strings; a set keeps its values empty and uses only the keys. */
typedef struct {
    kv_kind kind;
    union {
        kv_bytes str;
        kv_list list;
        kv_store *table; /* owned, for KV_HASH and KV_SET */
    } u;
} kv_value;

const char *kv_kind_name(kv_kind kind); /* "string", "list", "hash" or "set" */
void kv_value_free(kv_value *v);

/* A list: an array of owned byte strings. RPUSH appends; LPUSH shifts the array along by
 * one, which costs the length of the list. Redis keeps a linked structure instead, and
 * pays for that in other ways. */
typedef struct {
    kv_bytes *items; /* owned array of owned strings */
    size_t len;
    size_t cap;
} kv_list;

const char *kv_kind_name(kv_kind kind) {
    switch (kind) {
    case KV_STRING:
        return "string";
    case KV_LIST:
        return "list";
    case KV_HASH:
        return "hash";
    case KV_SET:
        return "set";
    }
    return "none";
}

bool kv_bytes_set(kv_bytes *b, const char *src, size_t n) {
    char *copy = kv_bytes_dup(src, n);
    if (copy == NULL) {
        return false;
    }
    free(b->p);
    b->p = copy;
    b->len = n;
    return true;
}

static void kv_list_free(kv_list *l) {
    for (size_t i = 0; i < l->len; i++) {
        free(l->items[i].p);
    }
    free(l->items);
}

bool kv_list_push(kv_list *l, const char *src, size_t n, bool front) {
    char *copy = kv_bytes_dup(src, n);
    if (copy == NULL) {
        return false;
    }
    if (l->len == l->cap) {
        size_t cap = l->cap ? l->cap * 2 : 4;
        kv_bytes *items = realloc(l->items, cap * sizeof *items);
        if (items == NULL) {
            free(copy);
            return false;
        }
        l->items = items;
        l->cap = cap;
    }
    if (front) {
        memmove(l->items + 1, l->items, l->len * sizeof *l->items);
        l->items[0] = (kv_bytes){copy, n};
    } else {
        l->items[l->len] = (kv_bytes){copy, n};
    }
    l->len++;
    return true;
}

/* Free whatever the tag says is live. The switch has no default, so a fifth kind would
 * be reported by the compiler here, with -Wall, until it is handled. */
void kv_value_free(kv_value *v) {
    switch (v->kind) {
    case KV_STRING:
        free(v->u.str.p);
        break;
    case KV_LIST:
        kv_list_free(&v->u.list);
        break;
    case KV_HASH:
    case KV_SET:
        kv_store_free(v->u.table);
        break;
    }
    *v = (kv_value){KV_STRING, {{NULL, 0}}};
}

#define KV_WRONGTYPE "WRONGTYPE Operation against a key holding the wrong kind of value"

/* The string of an entry, or NULL after writing the WRONGTYPE error. Every string
 * command starts here, so the check against the tag is in one place. */
static kv_bytes *kv_string_of(kv_entry *e, kv_buf *out) {
    if (e->val.kind != KV_STRING) {
        kv_reply_error(out, KV_WRONGTYPE);
        return NULL;
    }
    return &e->val.u.str;
}

/* The entry for key as the given kind, made empty when the key is new, or NULL after
 * writing the WRONGTYPE error when it holds another kind. */
static kv_entry *kv_entry_as(kv_store *s, const kv_cmd *c, kv_kind kind, kv_buf *out, bool *oom) {
    bool created;
    *oom = false;
    kv_entry *e = kv_store_insert(s, c->argv[1], c->arglen[1], &created);
    if (e == NULL) {
        *oom = true;
        return NULL;
    }
    if (created) {
        e->val.kind = kind;
        if (kind == KV_HASH || kind == KV_SET) {
            e->val.u.table = kv_store_new(kv_store_type_of(s));
            if (e->val.u.table == NULL) {
                kv_store_del(s, c->argv[1], c->arglen[1]);
                *oom = true;
                return NULL;
            }
        }
    } else if (e->val.kind != kind) {
        kv_reply_error(out, KV_WRONGTYPE);
        return NULL;
    }
    return e;
}

/* LPUSH and RPUSH: the same function, told which end by the table. */
static bool kv_push(const kv_cmd *c, kv_store *s, kv_buf *out, bool front) {
    bool oom;
    kv_entry *e = kv_entry_as(s, c, KV_LIST, out, &oom);
    if (e == NULL) {
        return !oom;
    }
    for (size_t i = 2; i < c->argc; i++) {
        if (!kv_list_push(&e->val.u.list, c->argv[i], c->arglen[i], front)) {
            return false;
        }
    }
    return kv_reply_int(out, (long long)e->val.u.list.len);
}

static bool kv_cmd_lpush(const kv_cmd *c, kv_store *s, kv_buf *out) {
    return kv_push(c, s, out, true);
}

static bool kv_cmd_rpush(const kv_cmd *c, kv_store *s, kv_buf *out) {
    return kv_push(c, s, out, false);
}

/* The list of an entry, or NULL after the WRONGTYPE error. */
static kv_list *kv_list_of(kv_entry *e, kv_buf *out) {
    if (e->val.kind != KV_LIST) {
        kv_reply_error(out, KV_WRONGTYPE);
        return NULL;
    }
    return &e->val.u.list;
}

static bool kv_cmd_llen(const kv_cmd *c, kv_store *s, kv_buf *out) {
    kv_entry *e = kv_store_find(s, c->argv[1], c->arglen[1]);
    if (e == NULL) {
        return kv_reply_int(out, 0);
    }
    kv_list *l = kv_list_of(e, out);
    return l == NULL || kv_reply_int(out, (long long)l->len);
}

static bool kv_cmd_lrange(const kv_cmd *c, kv_store *s, kv_buf *out) {
    long long start, end;
    if (!kv_parse_int(c->argv[2], c->arglen[2], &start) ||
        !kv_parse_int(c->argv[3], c->arglen[3], &end)) {
        return kv_reply_error(out, "ERR value is not an integer or out of range");
    }
    kv_entry *e = kv_store_find(s, c->argv[1], c->arglen[1]);
    if (e == NULL) {
        return kv_reply_array(out, 0);
    }
    kv_list *l = kv_list_of(e, out);
    if (l == NULL) {
        return true;
    }
    size_t from, to;
    if (!kv_clamp(start, end, (long long)l->len, &from, &to)) {
        return kv_reply_array(out, 0);
    }
    if (!kv_reply_array(out, to - from + 1)) {
        return false;
    }
    for (size_t i = from; i <= to; i++) {
        if (!kv_reply_bulk(out, l->items[i].p, l->items[i].len)) {
            return false;
        }
    }
    return true;
}

/* HSET key field value [field value ...]: the fields go into the nested store. */
static bool kv_cmd_hset(const kv_cmd *c, kv_store *s, kv_buf *out) {
    if (c->argc % 2 != 0) {
        return kv_reply_arity(out, "hset");
    }
    bool oom;
    kv_entry *e = kv_entry_as(s, c, KV_HASH, out, &oom);
    if (e == NULL) {
        return !oom;
    }
    long long added = 0;
    for (size_t i = 2; i + 1 < c->argc; i += 2) {
        added += kv_store_find(e->val.u.table, c->argv[i], c->arglen[i]) == NULL;
        if (!kv_store_set(e->val.u.table, c->argv[i], c->arglen[i], c->argv[i + 1], c->arglen[i + 1])) {
            return false;
        }
    }
    return kv_reply_int(out, added);
}

/* The nested table of an entry, or NULL after the WRONGTYPE error. */
static kv_store *kv_table_of(kv_entry *e, kv_kind kind, kv_buf *out) {
    if (e->val.kind != kind) {
        kv_reply_error(out, KV_WRONGTYPE);
        return NULL;
    }
    return e->val.u.table;
}

static bool kv_cmd_hget(const kv_cmd *c, kv_store *s, kv_buf *out) {
    kv_entry *e = kv_store_find(s, c->argv[1], c->arglen[1]);
    if (e == NULL) {
        return kv_reply_nil(out);
    }
    kv_store *h = kv_table_of(e, KV_HASH, out);
    if (h == NULL) {
        return true;
    }
    kv_entry *f = kv_store_find(h, c->argv[2], c->arglen[2]);
    return f ? kv_reply_bulk(out, f->val.u.str.p, f->val.u.str.len) : kv_reply_nil(out);
}

/* The context of the HGETALL and SMEMBERS walks: the reply so far, and whether a
 * reply failed to append. */
typedef struct {
    kv_buf *out;
    bool with_values;
    bool ok;
} kv_members_ctx;

static bool kv_members_visit(kv_entry *e, void *ctx) {
    kv_members_ctx *m = ctx;
    m->ok = kv_reply_bulk(m->out, e->key, e->klen) &&
            (!m->with_values || kv_reply_bulk(m->out, e->val.u.str.p, e->val.u.str.len));
    return m->ok;
}

static bool kv_cmd_hgetall(const kv_cmd *c, kv_store *s, kv_buf *out) {
    kv_entry *e = kv_store_find(s, c->argv[1], c->arglen[1]);
    if (e == NULL) {
        return kv_reply_array(out, 0);
    }
    kv_store *h = kv_table_of(e, KV_HASH, out);
    if (h == NULL) {
        return true;
    }
    if (!kv_reply_array(out, kv_store_len(h) * 2)) {
        return false;
    }
    kv_members_ctx m = {out, true, true};
    kv_store_each(h, kv_members_visit, &m);
    return m.ok;
}

size_t kv_store_scan(kv_store *s, size_t cursor, size_t count,
                     bool (*fn)(kv_entry *e, void *ctx), void *ctx) {
    size_t seen = 0;
    while (cursor < s->cap) {
        for (kv_entry *e = s->buckets[cursor]; e != NULL; e = e->next) {
            seen++;
            if (!fn(e, ctx)) {
                return 0;
            }
        }
        cursor++;
        if (seen >= count) {
            break;
        }
    }
    return cursor < s->cap ? cursor : 0;
}

static bool kv_cmd_type(const kv_cmd *c, kv_store *s, kv_buf *out) {
    kv_entry *e = kv_store_find(s, c->argv[1], c->arglen[1]);
    return kv_reply_simple(out, e ? kv_kind_name(e->val.kind) : "none");
}

/* The context of a SCAN walk: the pattern, and the matching keys collected so far. */
typedef struct {
    const char *pat;
    size_t plen;
    kv_buf keys;
    size_t n;
    bool ok;
} kv_scan_ctx;

static bool kv_scan_visit(kv_entry *e, void *ctx) {
    kv_scan_ctx *k = ctx;
    if (k->pat != NULL && !kv_glob(k->pat, k->plen, e->key, e->klen)) {
        return true;
    }
    if (!kv_reply_bulk(&k->keys, e->key, e->klen)) {
        k->ok = false;
        return false;
    }
    k->n++;
    return true;
}

/* SCAN cursor [MATCH pattern] [COUNT n]: a page of keys and the cursor for the next page.
 * The reply is an array of two things, the cursor as a string and the array of keys. */
static bool kv_cmd_scan(const kv_cmd *c, kv_store *s, kv_buf *out) {
    long long cursor;
    if (!kv_parse_int(c->argv[1], c->arglen[1], &cursor) || cursor < 0) {
        return kv_reply_error(out, "ERR invalid cursor");
    }
    kv_scan_ctx k = {NULL, 0, {0}, 0, true};
    long long count = 10;
    for (size_t i = 2; i + 1 < c->argc; i += 2) {
        if (kv_arg_is(c, i, "MATCH")) {
            k.pat = c->argv[i + 1];
            k.plen = c->arglen[i + 1];
        } else if (kv_arg_is(c, i, "COUNT")) {
            if (!kv_parse_int(c->argv[i + 1], c->arglen[i + 1], &count) || count < 1) {
                return kv_reply_error(out, "ERR value is not an integer or out of range");
            }
        } else {
            return kv_reply_error(out, "ERR syntax error");
        }
    }
    if (c->argc % 2 != 0) {
        return kv_reply_error(out, "ERR syntax error");
    }
    size_t next = kv_store_scan(s, (size_t)cursor, (size_t)count, kv_scan_visit, &k);
    char text[32];
    int len = snprintf(text, sizeof text, "%zu", next);
    bool ok = k.ok && kv_reply_array(out, 2) && kv_reply_bulk(out, text, (size_t)len) &&
              kv_reply_array(out, k.n) && kv_buf_append(out, k.keys.data, k.keys.len);
    kv_buf_free(&k.keys);
    return ok;
}

// A value is one of four things, and the store holds it as `any`: an interface with no
// methods, which any value satisfies. The four concrete types are the cases, and a type
// switch takes the value apart. ErrWrongType is what a command replies when the case is
// not the one it needs.
var ErrWrongType = errors.New("WRONGTYPE Operation against a key holding the wrong kind of value")

// Kind names the case, as TYPE reports it. The switch has no default on purpose: a
// value of a fifth type would reach "none", and the tests would say so.
func Kind(v any) string {
	switch v.(type) {
	case []byte:
		return "string"
	case *List:
		return "list"
	case Hash:
		return "hash"
	case *Set[string]:
		return "set"
	}
	return "none"
}

// stringOf is the value as bytes, or ErrWrongType. The assertion v.([]byte) asks whether
// the value inside the interface has that type; the second result says. Every string
// command starts here, so the check is in one place.
func stringOf(v any) ([]byte, error) {
	b, ok := v.([]byte)
	if !ok {
		return nil, store.ErrWrongType
	}
	return b, nil
}

// as returns the value under key as T, making one with mk when the key is new. A key
// that holds another kind is ErrWrongType. T is a type parameter, so one function
// serves *store.List, store.Hash and *store.Set[string].
func as[T any](s Storer, key []byte, mk func() T) (T, error) {
	if v, ok := s.Get(key); ok {
		t, isT := v.(T)
		if !isT {
			var zero T
			return zero, store.ErrWrongType
		}
		return t, nil
	}
	t := mk()
	s.Set(key, t)
	return t, nil
}

// A Hash is a map from field to value, with nothing to add: a map already does it all.
type Hash map[string][]byte

// A Set is a map with no values. It is generic, because nothing about it depends on
// what the members are, only that they can be compared.
type Set[T comparable] struct {
	m map[T]struct{}
}

func NewSet[T comparable]() *Set[T] { return &Set[T]{m: make(map[T]struct{})} }

// Add reports whether the member was new.
func (s *Set[T]) Add(v T) bool {
	_, had := s.m[v]
	s.m[v] = struct{}{}
	return !had
}

func (s *Set[T]) Has(v T) bool { _, ok := s.m[v]; return ok }
func (s *Set[T]) Len() int     { return len(s.m) }

// All yields every member, in no particular order.
func (s *Set[T]) All() iter.Seq[T] {
	return func(yield func(T) bool) {
		for v := range s.m {
			if !yield(v) {
				return
			}
		}
	}
}

// A List is a slice of byte strings. PushBack appends; PushFront makes room at the
// start by copying the slice along, which costs the length of the list.
type List struct {
	items [][]byte
}

func (l *List) PushBack(v []byte)  { l.items = append(l.items, v) }
func (l *List) PushFront(v []byte) { l.items = append([][]byte{v}, l.items...) }
func (l *List) Len() int           { return len(l.items) }

// Range yields the items from start to end, both inclusive, with negative offsets
// counted from the end and out-of-range offsets clamped. It is an iterator: a function
// that takes the loop body, and range calls it. The caller never sees the slice.
func (l *List) Range(start, end int64) iter.Seq[[]byte] {
	return func(yield func([]byte) bool) {
		n := int64(len(l.items))
		if start < 0 {
			start += n
		}
		if end < 0 {
			end += n
		}
		start = max(start, 0)
		end = min(end, n-1)
		for i := start; i <= end; i++ {
			if !yield(l.items[i]) {
				return
			}
		}
	}
}

// Scan returns a page of keys from the cursor on, and the cursor for the next page, or
// 0 when the walk is done. A Go map cannot be walked from a position: each range starts
// in a new random order. So the keys are sorted and the cursor is an index into that
// order, which costs a sort per page. The hand-written table scans by bucket instead.
func (s *Store) Scan(cursor uint64, count int) (uint64, [][]byte) {
	sorted := slices.Sorted(maps.Keys(s.m))
	if cursor >= uint64(len(sorted)) {
		return 0, nil
	}
	end := min(int(cursor)+count, len(sorted))
	var page [][]byte
	for _, k := range sorted[cursor:end] {
		page = append(page, []byte(k))
	}
	if end == len(sorted) {
		return 0, page
	}
	return uint64(end), page
}

// Scan walks the buckets from cursor, collecting keys until at least count have been
// seen or the buckets end. The returned cursor is the next bucket, or 0 at the end: a
// cursor is a bucket index, which is why the walk can resume.
func (t *Table[K, V]) Scan(cursor uint64, count int) (uint64, []K) {
	var page []K
	for cursor < uint64(len(t.buckets)) {
		for _, e := range t.buckets[cursor] {
			page = append(page, e.key)
		}
		cursor++
		if len(page) >= count {
			break
		}
	}
	if cursor >= uint64(len(t.buckets)) {
		return 0, page
	}
	return cursor, page
}

/// The four kinds of value, as one enum. Each variant carries its own data, and a match
/// must handle every variant, so a fifth kind added later is a compile error at every
/// place that forgot it. This is the sum type that C builds by hand with a tag and a
/// union, and Go builds with an interface and a type switch.
pub enum Value {
    Str(Vec<u8>),
    List(VecDeque<Vec<u8>>),
    Hash(HashMap<Vec<u8>, Vec<u8>>),
    Set(HashSet<Vec<u8>>),
}

impl Value {
    /// The name TYPE reports.
    pub fn kind(&self) -> &'static str {
        match self {
            Value::Str(_) => "string",
            Value::List(_) => "list",
            Value::Hash(_) => "hash",
            Value::Set(_) => "set",
        }
    }
}

/// The error a command replies when the value is not the kind it needs.
#[derive(Debug, PartialEq)]
pub struct WrongType;

impl WrongType {
    pub const MESSAGE: &'static str =
        "WRONGTYPE Operation against a key holding the wrong kind of value";
}

/// What the commands need from a store. Five methods work on values of any kind, and the
/// string commands are default methods written on top of them.
pub trait Store {
    fn value(&self, key: &[u8]) -> Option<&Value>;
    fn value_mut(&mut self, key: &[u8]) -> Option<&mut Value>;
    /// The value under key, made with `make` when the key is new.
    fn value_or_insert(&mut self, key: Vec<u8>, make: fn() -> Value) -> &mut Value;
    /// Store a value of any kind, replacing whatever kind was there.
    fn insert(&mut self, key: Vec<u8>, val: Value);
    fn del(&mut self, key: &[u8]) -> bool;
    fn len(&self) -> usize;
    /// Every key that matches the pattern, in no particular order.
    fn keys(&self, pattern: &[u8]) -> Vec<&[u8]>;
    /// A page of up to `count` keys from `cursor` on, and the cursor of the next page,
    /// or 0 when the walk is complete.
    fn scan(&self, cursor: usize, count: usize) -> (usize, Vec<&[u8]>);

    fn is_empty(&self) -> bool {
        self.len() == 0
    }

    /// The string under key: `None` for a missing key, an error for another kind.
    fn get(&self, key: &[u8]) -> Result<Option<&[u8]>, WrongType> {
        match self.value(key) {
            None => Ok(None),
            Some(Value::Str(s)) => Ok(Some(s)),
            Some(_) => Err(WrongType),
        }
    }

    fn set(&mut self, key: Vec<u8>, val: Vec<u8>) {
        self.insert(key, Value::Str(val));
    }

    /// Add one to the integer stored under key, which counts as 0 when absent.
    fn incr(&mut self, key: &[u8]) -> Result<i64, IncrError> {
        let n = match self.get(key)? {
            None => 0,
            Some(bytes) => parse_int(bytes).ok_or(IncrError::NotInteger)?,
        };
        let n = n.checked_add(1).ok_or(IncrError::Overflow)?;
        self.set(key.to_vec(), n.to_string().into_bytes());
        Ok(n)
    }
}

/// LPUSH and RPUSH: the same function, told which end by a flag. A `VecDeque` pushes at
/// either end in constant time, which is why the list is one and not a `Vec`.
fn push(
    args: &mut Command,
    store: &mut dyn Store,
    w: &mut dyn Write,
    front: bool,
) -> io::Result<()> {
    let key = mem::take(&mut args[1]);
    let Value::List(list) = store.value_or_insert(key, || Value::List(VecDeque::new())) else {
        return write_wrongtype(w);
    };
    for item in args.drain(2..) {
        if front {
            list.push_front(item);
        } else {
            list.push_back(item);
        }
    }
    resp::write_int(w, list.len() as i64)
}

fn lpush(args: &mut Command, store: &mut dyn Store, w: &mut dyn Write) -> io::Result<()> {
    push(args, store, w, true)
}

fn rpush(args: &mut Command, store: &mut dyn Store, w: &mut dyn Write) -> io::Result<()> {
    push(args, store, w, false)
}

/// The value under key as the kind the command needs: `None` for a missing key, an
/// error for another kind. `Kind<'a, T>` is a name for the shape all three helpers return.
type Kind<'a, T> = Result<Option<&'a T>, WrongType>;

/// The list under key, an empty one when the key is missing, or the wrong kind.
fn list_of<'a>(store: &'a dyn Store, key: &[u8]) -> Kind<'a, VecDeque<Vec<u8>>> {
    match store.value(key) {
        None => Ok(None),
        Some(Value::List(l)) => Ok(Some(l)),
        Some(_) => Err(WrongType),
    }
}

fn llen(args: &mut Command, store: &mut dyn Store, w: &mut dyn Write) -> io::Result<()> {
    match list_of(store, &args[1]) {
        Ok(l) => resp::write_int(w, l.map_or(0, VecDeque::len) as i64),
        Err(WrongType) => write_wrongtype(w),
    }
}

/// LRANGE key start end: a slice of the list by position. `range` on a `VecDeque` is
/// an iterator over the items in that range, and `write_bulks` drains it.
fn lrange(args: &mut Command, store: &mut dyn Store, w: &mut dyn Write) -> io::Result<()> {
    let (Some(start), Some(end)) = (parse_int(&args[2]), parse_int(&args[3])) else {
        return resp::write_error(w, "ERR value is not an integer or out of range");
    };
    let list = match list_of(store, &args[1]) {
        Ok(Some(l)) => l,
        Ok(None) => return resp::write_array_len(w, 0),
        Err(WrongType) => return write_wrongtype(w),
    };
    match clamp(start, end, list.len()) {
        Some(range) => write_bulks(w, list.range(range).map(Vec::as_slice)),
        None => resp::write_array_len(w, 0),
    }
}

/// A HashMap walks in one fixed order while it is not changed, so the cursor is a
/// count of keys already seen. `skip` and `take` are lazy: nothing runs until
/// `collect` asks, and the map is never copied. One more `next` tells whether a
/// page follows.
fn scan(&self, cursor: usize, count: usize) -> (usize, Vec<&[u8]>) {
    let mut rest = self.map.keys().skip(cursor);
    let page: Vec<&[u8]> = rest.by_ref().take(count).map(Vec::as_slice).collect();
    let next = if rest.next().is_some() {
        cursor + page.len()
    } else {
        0
    };
    (next, page)
}

fn sadd(args: &mut Command, store: &mut dyn Store, w: &mut dyn Write) -> io::Result<()> {
    let key = mem::take(&mut args[1]);
    let Value::Set(set) = store.value_or_insert(key, || Value::Set(HashSet::new())) else {
        return write_wrongtype(w);
    };
    let added = args.drain(2..).filter(|m| set.insert(m.clone())).count();
    resp::write_int(w, added as i64)
}

fn set_of<'a>(store: &'a dyn Store, key: &[u8]) -> Kind<'a, HashSet<Vec<u8>>> {
    match store.value(key) {
        None => Ok(None),
        Some(Value::Set(s)) => Ok(Some(s)),
        Some(_) => Err(WrongType),
    }
}

fn sismember(args: &mut Command, store: &mut dyn Store, w: &mut dyn Write) -> io::Result<()> {
    match set_of(store, &args[1]) {
        Ok(s) => resp::write_int(w, i64::from(s.is_some_and(|s| s.contains(&args[2])))),
        Err(WrongType) => write_wrongtype(w),
    }
}

fn smembers(args: &mut Command, store: &mut dyn Store, w: &mut dyn Write) -> io::Result<()> {
    match set_of(store, &args[1]) {
        Ok(Some(s)) => write_bulks(w, s.iter().map(Vec::as_slice)),
        Ok(None) => resp::write_array_len(w, 0),
        Err(WrongType) => write_wrongtype(w),
    }
}

/// A recursive enum. Without the `Box`, a `Chain` would contain a `Chain` directly, and
/// the compiler could not say how large one is. `Box` puts the next link on the heap,
/// which makes the size one pointer, and it owns that link, so dropping the head drops
/// the whole chain.
pub enum Chain {
    Link(Vec<u8>, Box<Chain>),
    End,
}

impl Chain {
    pub fn push(self, item: Vec<u8>) -> Chain {
        Chain::Link(item, Box::new(self))
    }
}

/// An iterator over a chain: a value with one method, `next`, which gives the next item
/// or `None`. Everything else, `map`, `filter`, `count` and the `for` loop, comes with
/// the trait. The iterator takes the chain apart as it goes, so it owns it.
pub struct IntoIter(Chain);

impl Iterator for IntoIter {
    type Item = Vec<u8>;

    fn next(&mut self) -> Option<Vec<u8>> {
        match std::mem::replace(&mut self.0, Chain::End) {
            Chain::Link(item, rest) => {
                self.0 = *rest;
                Some(item)
            }
            Chain::End => None,
        }
    }
}

impl IntoIterator for Chain {
    type Item = Vec<u8>;
    type IntoIter = IntoIter;

    fn into_iter(self) -> IntoIter {
        IntoIter(self)
    }
}

/* One client: its socket, the bytes read so far, and the replies not yet written. */
typedef struct {
    int fd;
    kv_buf in;
    kv_buf out;
    bool writing; /* the poller is watching for room to write */
    bool closing; /* after a protocol error: close once out is written */
    bool dead;    /* gone; removed once the batch of events it was part of is done */
} kv_conn;

/* Every client, beside the poller that watches their sockets. The array holds
 * pointers, so a client stays where it is in memory when another one is removed, and
 * the tag the poller hands back stays valid. */
typedef struct {
    kv_conn **conns;
    size_t len;
    size_t cap;
    kv_poller *poller;
} kv_clients;

/* A non-blocking socket returns EAGAIN instead of waiting, which the loop relies on. */
static bool kv_set_nonblocking(int fd) {
    int flags = fcntl(fd, F_GETFL, 0);
    return flags >= 0 && fcntl(fd, F_SETFL, flags | O_NONBLOCK) >= 0;
}

/* Accept every waiting client. The listening socket is non-blocking, so accept()
 * returns EAGAIN when the queue is empty. */
static void kv_accept_all(int listen_fd, kv_clients *cs) {
    for (;;) {
        int fd = accept(listen_fd, NULL, NULL);
        if (fd < 0) {
            if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) {
                return;
            }
            perror("accept");
            return;
        }
        if (!kv_set_nonblocking(fd) || !kv_clients_add(cs, fd)) {
            close(fd);
        }
    }
}

/* Read what has arrived, run every complete request, and queue the replies. Returns
 * false when the client has gone or must be dropped. */
static bool kv_conn_input(kv_conn *c, kv_store *store) {
    char tmp[4096];
    for (;;) {
        ssize_t n = read(c->fd, tmp, sizeof tmp);
        if (n < 0) {
            if (errno == EAGAIN || errno == EWOULDBLOCK) {
                break; /* nothing more for now */
            }
            if (errno == EINTR) {
                continue;
            }
            return false;
        }
        if (n == 0) {
            return false; /* the client closed */
        }
        if (!kv_buf_append(&c->in, tmp, (size_t)n)) {
            return false;
        }
    }
    for (;;) {
        kv_cmd cmd;
        size_t used = 0;
        kv_parse_status st = kv_parse(c->in.data, c->in.len, &cmd, &used);
        if (st == KV_PARSE_INCOMPLETE && c->in.len > KV_MAX_REQUEST) {
            st = KV_PARSE_BAD; /* a request that never ends is not a request */
        }
        /* No default: with -Wall the compiler reports a status this switch forgot. */
        switch (st) {
        case KV_PARSE_INCOMPLETE:
            return true; /* the poller says when more bytes arrive */
        case KV_PARSE_BAD:
            kv_reply_error(&c->out, "ERR Protocol error");
            c->closing = true;
            return true;
        case KV_PARSE_OK:
            if (!kv_exec(&cmd, store, &c->out)) {
                return false;
            }
            kv_buf_consume(&c->in, used);
            break;
        }
    }
}

/* One client had an event. Read if there is input, write if there is room or a reply is
 * queued, then ask to be told about room only while a reply is waiting. A socket is
 * almost always writable, so asking every time would make every wait return at once. */
static void kv_conn_ready(kv_clients *cs, kv_conn *c, const kv_event *ev, kv_store *store) {
    bool alive = true;
    if (ev->readable || ev->hangup) {
        alive = kv_conn_input(c, store);
    }
    if (alive && (ev->writable || c->out.len > 0)) {
        alive = kv_conn_output(c);
    }
    if (!alive || (c->closing && c->out.len == 0)) {
        c->dead = true; /* not freed yet: this batch may hold another event for it */
        return;
    }
    bool want = c->out.len > 0;
    if (want != c->writing && kv_poller_want_write(cs->poller, c->fd, c, want)) {
        c->writing = want;
    }
}

int kv_server_run(int listen_fd, kv_store *store) {
    kv_clients cs = {0};
    cs.poller = kv_poller_new();
    /* the listening socket is watched too, with a null tag to tell it from a client */
    if (cs.poller == NULL || !kv_set_nonblocking(listen_fd) ||
        !kv_poller_add(cs.poller, listen_fd, NULL)) {
        kv_poller_free(cs.poller);
        return 1;
    }
    fprintf(stderr, "kv waits with %s\n", kv_poller_name());
    kv_event events[64];
    for (;;) {
        int n = kv_poller_wait(cs.poller, events, 64, -1);
        if (n < 0) {
            if (errno == EINTR) {
                continue;
            }
            perror("wait");
            break;
        }
        for (int i = 0; i < n; i++) {
            kv_conn *c = events[i].tag;
            if (c == NULL) {
                kv_accept_all(listen_fd, &cs);
            } else if (!c->dead) {
                kv_conn_ready(&cs, c, &events[i], store);
            }
        }
        /* walk backwards, because removing a client moves the last one into its slot */
        for (size_t i = cs.len; i-- > 0;) {
            if (cs.conns[i]->dead) {
                kv_clients_remove(&cs, i);
            }
        }
    }
    for (size_t i = cs.len; i-- > 0;) {
        kv_clients_remove(&cs, i);
    }
    free(cs.conns);
    kv_poller_free(cs.poller);
    return 1;
}

/* Every client is an open file, and a process starts with a soft limit on those, often
 * 256 on macOS and 1024 on Linux. Raise it to the hard limit, as Redis does, and say
 * how many clients that leaves room for. */
static void kv_raise_fd_limit(void) {
    struct rlimit lim;
    if (getrlimit(RLIMIT_NOFILE, &lim) < 0) {
        return;
    }
    if (lim.rlim_cur < lim.rlim_max && setrlimit(RLIMIT_NOFILE, &(struct rlimit){lim.rlim_max, lim.rlim_max}) == 0) {
        lim.rlim_cur = lim.rlim_max;
    }
    if (lim.rlim_cur == RLIM_INFINITY) {
        fprintf(stderr, "kv has no limit on open files\n");
    } else {
        /* the listener, the poller, stdio and some spare take the rest */
        fprintf(stderr, "kv can hold %llu clients\n", (unsigned long long)lim.rlim_cur - 32);
    }
}

/* One of poller_poll.c, poller_kqueue.c and poller_epoll.c holds the code; the other two
 * compile to nothing. The compiler predefines a macro for the target: __linux__ on
 * Linux, __APPLE__ on macOS, __FreeBSD__ on FreeBSD. KV_POLL on the command line, as in
 * `make kv KV_POLL=1`, forces the portable poll() version on any system. */
#if defined(KV_POLL)
#define KV_POLLER_POLL 1
#elif defined(__linux__)
#define KV_POLLER_EPOLL 1
#elif defined(__APPLE__) || defined(__FreeBSD__)
#define KV_POLLER_KQUEUE 1
#else
#define KV_POLLER_POLL 1
#endif

typedef struct kv_poller kv_poller;

/* What a wait reports for one socket: the tag given when it was added, and which of
 * the three things happened to it. */
typedef struct {
    void *tag;
    bool readable;
    bool writable;
    bool hangup;
} kv_event;

kv_poller *kv_poller_new(void);
void kv_poller_free(kv_poller *p);

/* Watch fd for input. The tag comes back in every event for fd. */
bool kv_poller_add(kv_poller *p, int fd, void *tag);

/* Also watch fd for room to write, or stop watching for it. */
bool kv_poller_want_write(kv_poller *p, int fd, void *tag, bool want);

/* Stop watching fd. Call it before close(). */
void kv_poller_remove(kv_poller *p, int fd);

/* Block until a socket is ready or timeout_ms passes, where -1 means wait for ever.
 * Fills up to max events and returns how many, 0 on timeout, or -1 with errno set. */
int kv_poller_wait(kv_poller *p, kv_event *events, int max, int timeout_ms);

/* "poll", "kqueue" or "epoll", for the startup line. */
const char *kv_poller_name(void);

/* The whole state is one descriptor: the kernel holds the list. */
struct kv_poller {
    int ep;
};

kv_poller *kv_poller_new(void) {
    kv_poller *p = malloc(sizeof *p);
    if (p == NULL) {
        return NULL;
    }
    p->ep = epoll_create1(0);
    if (p->ep < 0) {
        free(p);
        return NULL;
    }
    return p;
}

void kv_poller_free(kv_poller *p) {
    if (p != NULL) {
        close(p->ep);
        free(p);
    }
}

/* One socket has one entry with one mask of events, so a change replaces the mask. */
static bool kv_ep_ctl(kv_poller *p, int op, int fd, uint32_t mask, void *tag) {
    struct epoll_event change = {.events = mask, .data.ptr = tag};
    return epoll_ctl(p->ep, op, fd, &change) == 0;
}

bool kv_poller_add(kv_poller *p, int fd, void *tag) {
    return kv_ep_ctl(p, EPOLL_CTL_ADD, fd, EPOLLIN, tag);
}

bool kv_poller_want_write(kv_poller *p, int fd, void *tag, bool want) {
    return kv_ep_ctl(p, EPOLL_CTL_MOD, fd, EPOLLIN | (want ? EPOLLOUT : 0), tag);
}

/* close() drops a socket from the list by itself, unless another descriptor refers to
 * the same socket, which is why the explicit call is the safe habit. */
void kv_poller_remove(kv_poller *p, int fd) {
    epoll_ctl(p->ep, EPOLL_CTL_DEL, fd, NULL);
}

int kv_poller_wait(kv_poller *p, kv_event *events, int max, int timeout_ms) {
    struct epoll_event ready[64];
    if (max > 64) {
        max = 64;
    }
    int n = epoll_wait(p->ep, ready, max, timeout_ms);
    if (n <= 0) {
        return n;
    }
    for (int i = 0; i < n; i++) {
        uint32_t mask = ready[i].events;
        events[i] = (kv_event){
            .tag = ready[i].data.ptr,
            .readable = (mask & EPOLLIN) != 0,
            .writable = (mask & EPOLLOUT) != 0,
            .hangup = (mask & (EPOLLHUP | EPOLLERR)) != 0,
        };
    }
    return n;
}

const char *kv_poller_name(void) {
    return "epoll";
}

func main() {
	ln, err := net.Listen("tcp", ":"+port())
	if err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(1)
	}
	fmt.Fprintln(os.Stderr, "kv listening on port", port())
	if n, ok := sys.OpenFiles(); ok {
		fmt.Fprintln(os.Stderr, "kv can hold", n-32, "clients")
	}
	s := store.New()
	for {
		conn, err := ln.Accept()
		if err != nil {
			fmt.Fprintln(os.Stderr, "accept:", err)
			continue
		}
		// the one word that changes: the client is served on its own goroutine, and
		// the loop goes straight back to accept
		go func() {
			if err := serve(conn, s); err != nil {
				fmt.Fprintln(os.Stderr, "client:", err)
			}
		}()
	}
}

// Store keeps every key and its value in a map, behind a lock. Many goroutines read and
// write the store at once, and a Go map is not safe for that without one. The value is
// any of the four kinds in value.go.
type Store struct {
	mu sync.RWMutex
	m  map[string]any
}

func New() *Store {
	return &Store{m: make(map[string]any)}
}

// Get takes the read lock, which any number of readers may hold together.
func (s *Store) Get(key []byte) (any, bool) {
	s.mu.RLock()
	defer s.mu.RUnlock()
	v, ok := s.m[string(key)]
	return v, ok
}

// Set takes the write lock, which excludes every other reader and writer.
func (s *Store) Set(key []byte, val any) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.m[string(key)] = val
}

func (s *Store) Del(key []byte) bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	_, ok := s.m[string(key)]
	delete(s.m, string(key))
	return ok
}

func (s *Store) Len() int {
	s.mu.RLock()
	defer s.mu.RUnlock()
	return len(s.m)
}

// Keys returns every key that matches the pattern, in no particular order.
func (s *Store) Keys(pattern []byte) [][]byte {
	s.mu.RLock()
	defer s.mu.RUnlock()
	var out [][]byte
	for k := range s.m {
		if glob(pattern, []byte(k)) {
			out = append(out, []byte(k))
		}
	}
	return out
}

// Scan returns a page of keys from the cursor on, in sorted order, and the cursor for
// the next page, or 0 when the walk is done.
func (s *Store) Scan(cursor uint64, count int) (uint64, [][]byte) {
	s.mu.RLock()
	defer s.mu.RUnlock()
	sorted := slices.Sorted(maps.Keys(s.m))
	if cursor >= uint64(len(sorted)) {
		return 0, nil
	}
	end := min(int(cursor)+count, len(sorted))
	var page [][]byte
	for _, k := range sorted[cursor:end] {
		page = append(page, []byte(k))
	}
	if end == len(sorted) {
		return 0, page
	}
	return uint64(end), page
}

// TestIncrFromManyGoroutines adds one to the same key from 50 goroutines, 200 times
// each. With the lock the result is 10000. Without it, go test -race reports the data
// race, and the count comes out short.
func TestIncrFromManyGoroutines(t *testing.T) {
	s := New()
	var wg sync.WaitGroup
	for range 50 {
		wg.Go(func() {
			for range 200 {
				if _, err := s.Incr([]byte("n")); err != nil {
					t.Error(err)
				}
			}
		})
	}
	wg.Wait()
	got, _ := s.Get([]byte("n"))
	if b, _ := got.([]byte); string(b) != "10000" {
		t.Fatalf("n = %s, want 10000", got)
	}
}

// answer is one round trip: which connection made it and how long it took.
type answer struct {
	worker int
	took   time.Duration
}

// worker owns one connection. It takes job after job from the channel until the channel
// is closed and empty, and reports each round trip on answers, or its first error on
// errs, after which it gives up.
func worker(id int, addr string, jobs <-chan int, answers chan<- answer, errs chan<- error) {
	conn, err := net.Dial("tcp", addr)
	if err != nil {
		errs <- fmt.Errorf("worker %d: %w", id, err)
		return
	}
	defer conn.Close()
	r := bufio.NewReader(conn)
	for range jobs {
		start := time.Now()
		if _, err := conn.Write([]byte("PING\r\n")); err != nil {
			errs <- fmt.Errorf("worker %d: %w", id, err)
			return
		}
		if _, err := r.ReadString('\n'); err != nil {
			errs <- fmt.Errorf("worker %d: %w", id, err)
			return
		}
		answers <- answer{id, time.Since(start)}
	}
}

func main() {
	addr := flag.String("addr", "127.0.0.1:6380", "the server")
	clients := flag.Int("c", 50, "connections")
	n := flag.Int("n", 10000, "requests in total")
	timeout := flag.Duration("timeout", 5*time.Second, "give up after this long")
	flag.Parse()

	// A buffered channel holds every job up front. Closing it tells the workers that
	// nothing more is coming: their range loops end once the buffer is empty.
	jobs := make(chan int, *n)
	for i := range *n {
		jobs <- i
	}
	close(jobs)

	answers := make(chan answer, *n)
	errs := make(chan error, *clients)
	var wg sync.WaitGroup
	for id := range *clients {
		wg.Go(func() { worker(id, *addr, jobs, answers, errs) })
	}
	// Only the sender may close a channel, and here there are many senders, so a
	// goroutine of its own waits for all of them and closes both channels after.
	go func() {
		wg.Wait()
		close(answers)
		close(errs)
	}()

	start := time.Now()
	var took []time.Duration
	byWorker := map[int]int{}
	failed := 0
	deadline := time.After(*timeout)
	// select waits on three channels at once. A closed channel is set to nil, and a nil
	// channel never fires, which takes its case out of the select until the loop ends.
	for answers != nil || errs != nil {
		select {
		case a, ok := <-answers:
			if !ok {
				answers = nil
				continue
			}
			took = append(took, a.took)
			byWorker[a.worker]++
		case err, ok := <-errs:
			if !ok {
				errs = nil
				continue
			}
			failed++
			fmt.Fprintln(os.Stderr, err)
		case <-deadline:
			// the workers still blocked on a read are left behind: the process is
			// about to exit, so their goroutines go with it
			fmt.Printf("gave up after %v\n", *timeout)
			answers, errs = nil, nil
		}
	}
	report(took, byWorker, *clients, failed, time.Since(start))
}

// OpenFiles reports how many files, and so how many clients, the process may hold open.
// The Go runtime has already raised the soft limit to the hard one at startup.
func OpenFiles() (uint64, bool) {
	var lim syscall.Rlimit
	if err := syscall.Getrlimit(syscall.RLIMIT_NOFILE, &lim); err != nil {
		return 0, false
	}
	return lim.Cur, true
}

// OpenFiles reports false where the system has no such limit to ask about.
func OpenFiles() (uint64, bool) {
	return 0, false
}

fn main() -> io::Result<()> {
    let port = port();
    let listener = TcpListener::bind(format!("0.0.0.0:{port}"))?;
    eprintln!("kv listening on port {port}");
    if let Some(n) = kv::sys::open_files() {
        eprintln!("kv can hold {} clients", n - 32);
    }
    let store = Arc::new(Mutex::new(MapStore::new()));
    for stream in listener.incoming() {
        match stream {
            Ok(stream) => {
                // a clone of the Arc is a new owner of the same store, moved into the thread
                let store = Arc::clone(&store);
                thread::spawn(move || {
                    if let Err(e) = serve(stream, store) {
                        eprintln!("client: {e}");
                    }
                });
            }
            Err(e) => eprintln!("accept: {e}"),
        }
    }
    Ok(())
}

/// The store is shared: an Arc counts the owners, and a Mutex lets one thread at a
/// time use it. The lock is taken for one command and released at the end of the line.
fn serve(stream: TcpStream, store: Arc<Mutex<MapStore>>) -> resp::Result<()> {
    let mut reader = BufReader::new(stream.try_clone()?);
    let mut writer = BufWriter::new(stream);
    loop {
        match resp::read_command(&mut reader) {
            Ok(Some(args)) => {
                let mut store = store.lock().unwrap_or_else(|e| e.into_inner());
                cmd::execute(args, &mut *store, &mut writer)?;
            }
            Ok(None) => return Ok(()),
            Err(e @ resp::KvError::Protocol(_)) => {
                resp::write_error(&mut writer, "ERR Protocol error")?;
                writer.flush()?;
                return Err(e);
            }
            Err(e) => return Err(e),
        }
        if reader.buffer().is_empty() {
            writer.flush()?;
        }
    }
}

/// One connection. It takes job numbers from the shared counter until they run out and
/// sends each round trip on the channel. An error ends the worker, and the caller
/// reports it. The read timeout is what lets a worker give up on a server that never
/// answers, because a scoped thread must finish before the scope ends.
fn worker(
    id: usize,
    addr: SocketAddr,
    jobs: &AtomicUsize,
    total: usize,
    timeout: Duration,
    answers: &mpsc::Sender<Answer>,
) -> io::Result<()> {
    let mut stream = TcpStream::connect_timeout(&addr, timeout)?;
    stream.set_read_timeout(Some(timeout))?;
    let mut reader = BufReader::new(stream.try_clone()?);
    let mut line = String::new();
    // fetch_add returns the old value, so every job number is handed out exactly once
    while jobs.fetch_add(1, Ordering::Relaxed) < total {
        let start = Instant::now();
        stream.write_all(b"PING\r\n")?;
        line.clear();
        if reader.read_line(&mut line)? == 0 {
            return Err(io::ErrorKind::UnexpectedEof.into());
        }
        // send fails only when main has stopped listening, and then there is nothing to do
        let _ = answers.send(Answer {
            worker: id,
            took: start.elapsed(),
        });
    }
    Ok(())
}

fn main() -> io::Result<()> {
    let (addr, clients, total, timeout) = args();
    let jobs = AtomicUsize::new(0);
    let (tx, rx) = mpsc::channel();
    let failed = AtomicUsize::new(0);
    let start = Instant::now();
    // The scope borrows jobs, failed and tx from this stack frame, and the threads it
    // spawns can use them without an Arc, because the scope joins every thread before
    // it returns. No thread outlives the data.
    let mut took: Vec<Duration> = thread::scope(|s| {
        for id in 0..clients {
            let tx = tx.clone();
            let (jobs, failed) = (&jobs, &failed);
            s.spawn(move || {
                if let Err(e) = worker(id, addr, jobs, total, timeout, &tx) {
                    failed.fetch_add(1, Ordering::Relaxed);
                    eprintln!("worker {id}: {e}");
                }
            });
        }
        // Each worker holds a clone of the sender. Once main drops its own, the receiver
        // ends when the last worker drops the last clone, which ends the for loop.
        drop(tx);
        let mut took = Vec::new();
        let mut by_worker = vec![0usize; clients];
        for answer in rx {
            took.push(answer.took);
            by_worker[answer.worker] += 1;
        }
        let answered = by_worker.iter().filter(|n| **n > 0).count();
        let most = by_worker.iter().max().copied().unwrap_or(0);
        println!(
            "{} answered by {answered} of {clients} connections, {most} of them by one, {} failed, {:.2} s in all",
            took.len(),
            failed.load(Ordering::Relaxed),
            start.elapsed().as_secs_f64()
        );
        took
    });
    if !took.is_empty() {
        took.sort();
        let ms = |d: Duration| d.as_secs_f64() * 1e3;
        println!(
            "p50 {:.3} ms, max {:.3} ms",
            ms(took[took.len() / 2]),
            ms(took[took.len() - 1])
        );
    }
    Ok(())
}

/// How many files, and so how many clients, the process may hold open. The standard
/// library has no call for it, so this asks the C library, which chapter 15 explains.
#[cfg(unix)]
pub fn open_files() -> Option<u64> {
    let mut lim = libc::rlimit {
        rlim_cur: 0,
        rlim_max: 0,
    };
    // SAFETY: getrlimit writes one struct through a valid pointer and reads nothing.
    let rc = unsafe { libc::getrlimit(libc::RLIMIT_NOFILE, &mut lim) };
    (rc == 0 && lim.rlim_cur != libc::RLIM_INFINITY).then_some(lim.rlim_cur)
}

/// Where the system has no such limit to ask about, the answer is None.
#[cfg(not(unix))]
pub fn open_files() -> Option<u64> {
    None
}

/// The replies are written into a Vec first, with the ordinary synchronous command
/// code from the library, and then sent with one awaited write. The lock is never held
/// across an await, so a plain std Mutex is the right one.
async fn serve(stream: TcpStream, store: Arc<Mutex<MapStore>>) -> resp::Result<()> {
    let mut reader = BufReader::new(stream);
    let mut out = Vec::new();
    loop {
        match read_command(&mut reader).await {
            Ok(Some(args)) => {
                let mut store = store.lock().unwrap_or_else(|e| e.into_inner());
                cmd::execute(args, &mut *store, &mut out)?;
            }
            Ok(None) => return Ok(()),
            Err(e @ KvError::Protocol(_)) => {
                resp::write_error(&mut out, "ERR Protocol error")?;
                reader.get_mut().write_all(&out).await?;
                return Err(e);
            }
            Err(e) => return Err(e),
        }
        if reader.buffer().is_empty() {
            reader.get_mut().write_all(&out).await?;
            out.clear();
        }
    }
}

/// The parser from chapter 4 again, with `.await` after every read. A blocking read
/// would stop the whole thread; an awaited one lets the thread run another task.
async fn read_line(r: &mut BufReader<TcpStream>) -> resp::Result<Option<Vec<u8>>> {
    let mut line = Vec::new();
    if r.read_until(b'\n', &mut line).await? == 0 {
        return Ok(None);
    }
    if line.pop() != Some(b'\n') {
        return Err(io::Error::from(io::ErrorKind::UnexpectedEof).into());
    }
    if line.last() == Some(&b'\r') {
        line.pop();
    }
    Ok(Some(line))
}

fn parse_len(digits: &[u8], type_byte: u8) -> resp::Result<usize> {
    std::str::from_utf8(digits)
        .ok()
        .and_then(|s| s.parse().ok())
        .ok_or(KvError::Protocol(type_byte))
}

async fn read_command(r: &mut BufReader<TcpStream>) -> resp::Result<Option<Command>> {
    let Some(line) = read_line(r).await? else {
        return Ok(None);
    };
    if line.first() != Some(&b'*') {
        let words = line.split(|b| *b == b' ').filter(|w| !w.is_empty());
        return Ok(Some(words.map(<[u8]>::to_vec).collect()));
    }
    let n = parse_len(&line[1..], b'*')?;
    if n > MAX_ARGS {
        return Err(KvError::Protocol(b'*'));
    }
    let mut args = Vec::with_capacity(n);
    for _ in 0..n {
        let line = read_line(r)
            .await?
            .ok_or(io::Error::from(io::ErrorKind::UnexpectedEof))?;
        if line.first() != Some(&b'$') {
            return Err(KvError::Protocol(line.first().copied().unwrap_or(b'\n')));
        }
        let size = parse_len(&line[1..], b'$')?;
        if size > MAX_BULK {
            return Err(KvError::Protocol(b'$'));
        }
        let mut buf = vec![0u8; size + 2];
        r.read_exact(&mut buf).await?;
        if &buf[size..] != b"\r\n" {
            return Err(KvError::Protocol(buf[size]));
        }
        buf.truncate(size);
        args.push(buf);
    }
    Ok(Some(args))
}

/* Milliseconds on the monotonic clock, which only moves forward. It is not the time of
 * day: that clock jumps when the system time is set, and a deadline must not. */
int64_t kv_now_ms(void);

/* One key and its value. expires_at is a deadline in milliseconds on the monotonic
 * clock, or 0 for a key that never expires. The key, the value and the next entry are
 * owned. */
typedef struct kv_entry {
    char *key;
    size_t klen;
    kv_value val;
    int64_t expires_at;
    struct kv_entry *next;
} kv_entry;

static bool kv_expired(const kv_entry *e, int64_t now) {
    return e->expires_at != 0 && e->expires_at <= now;
}

/* Unlink and free the entry that *link points to. */
static void kv_unlink(kv_store *s, kv_entry **link) {
    kv_entry *e = *link;
    *link = e->next;
    kv_entry_free(e);
    s->len--;
}

kv_entry *kv_store_find(kv_store *s, const char *key, size_t klen, int64_t now) {
    if (s->cap == 0) {
        return NULL;
    }
    size_t i = kv_index(s, key, klen, s->cap);
    for (kv_entry **link = &s->buckets[i]; *link != NULL; link = &(*link)->next) {
        kv_entry *e = *link;
        if (s->type->eq(e->key, e->klen, key, klen)) {
            if (kv_expired(e, now)) {
                kv_unlink(s, link); /* lazy expiry: the key dies when it is touched */
                return NULL;
            }
            return e;
        }
    }
    return NULL;
}

/* Wait at most 100 ms, so that keys expire even when no client is talking. */
int n = kv_poller_wait(cs.poller, events, 64, 100);
if (n < 0) {
    if (errno == EINTR) {
        continue;
    }
    perror("wait");
    break;
}
kv_store_sweep(store, kv_now_ms(), 200);

/* SET key value [EX seconds | PX milliseconds]. A plain SET removes any deadline, as
 * Redis does; a SET with EX or PX sets a new one. */
static bool kv_cmd_set(const kv_cmd *c, kv_store *s, kv_buf *out, int64_t now) {
    if (c->argc != 3 && c->argc != 5) {
        return kv_reply_arity(out, "set");
    }
    int64_t ttl_ms = 0;
    if (c->argc == 5) {
        long long n;
        bool ex = kv_arg_is(c, 3, "EX");
        if (!ex && !kv_arg_is(c, 3, "PX")) {
            return kv_reply_error(out, "ERR syntax error");
        }
        if (!kv_parse_int(c->argv[4], c->arglen[4], &n)) {
            return kv_reply_error(out, "ERR value is not an integer or out of range");
        }
        if (n <= 0 || n > INT64_MAX / 1000) {
            return kv_reply_error(out, "ERR invalid expire time in 'set' command");
        }
        ttl_ms = ex ? n * 1000 : n;
    }
    if (!kv_store_set(s, c->argv[1], c->arglen[1], c->argv[2], c->arglen[2])) {
        return false;
    }
    kv_entry *e = kv_store_find(s, c->argv[1], c->arglen[1], 0);
    e->expires_at = ttl_ms ? now + ttl_ms : 0;
    return kv_reply_simple(out, "OK");
}

/* EXPIRE key seconds: 1 if the key exists and got a deadline, 0 if it does not exist. */
static bool kv_cmd_expire(const kv_cmd *c, kv_store *s, kv_buf *out, int64_t now) {
    long long secs;
    if (!kv_parse_int(c->argv[2], c->arglen[2], &secs)) {
        return kv_reply_error(out, "ERR value is not an integer or out of range");
    }
    if (secs > INT64_MAX / 1000) { /* now + secs * 1000 must not overflow */
        return kv_reply_error(out, "ERR invalid expire time in 'expire' command");
    }
    kv_entry *e = kv_store_find(s, c->argv[1], c->arglen[1], now);
    if (e == NULL) {
        return kv_reply_int(out, 0);
    }
    if (secs <= 0) {
        kv_store_del(s, c->argv[1], c->arglen[1]); /* a deadline in the past is a DEL */
    } else {
        e->expires_at = now + secs * 1000;
    }
    return kv_reply_int(out, 1);
}

/* TTL key: the seconds left, -1 for a key with no deadline, -2 for a missing key. */
static bool kv_cmd_ttl(const kv_cmd *c, kv_store *s, kv_buf *out, int64_t now) {
    kv_entry *e = kv_store_find(s, c->argv[1], c->arglen[1], now);
    if (e == NULL) {
        return kv_reply_int(out, -2);
    }
    if (e->expires_at == 0) {
        return kv_reply_int(out, -1);
    }
    return kv_reply_int(out, (e->expires_at - now + 500) / 1000); /* nearest, as Redis */
}

// entry is a value of any kind and its deadline. A zero time means the key never
// expires. The deadline is a time.Time, which carries a monotonic reading, so a change
// to the system clock does not move it.
type entry struct {
	val     any
	expires time.Time
}

func (e entry) expired(now time.Time) bool {
	return !e.expires.IsZero() && !now.Before(e.expires)
}

// TestExpiry uses a clock the test controls, so it never sleeps.
func TestExpiry(t *testing.T) {
	s := New()
	t0 := time.Now()
	s.Set([]byte("k"), []byte("v"))
	s.SetExpiry([]byte("k"), t0.Add(time.Second), t0)
	if _, ok := s.Get([]byte("k"), t0.Add(999*time.Millisecond)); !ok {
		t.Fatal("key died early")
	}
	if _, ok := s.Get([]byte("k"), t0.Add(time.Second)); ok {
		t.Fatal("key outlived its deadline")
	}
	if n := s.Sweep(t0.Add(time.Second), 100); n != 1 {
		t.Fatalf("sweep removed %d keys, want 1", n)
	}
	if s.Len() != 0 {
		t.Fatalf("store holds %d keys after the sweep, want 0", s.Len())
	}
}

// SetExpiry gives a live key a deadline. It reports whether the key was there.
func (s *Store) SetExpiry(key []byte, at time.Time, now time.Time) bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	e, ok := s.m[string(key)]
	if !ok || e.expired(now) {
		return false
	}
	e.expires = at
	s.m[string(key)] = e
	return true
}

// TTL returns the time left, and a status: -2 for a missing key, -1 for a key with
// no deadline, 0 for a key with one.
func (s *Store) TTL(key []byte, now time.Time) (time.Duration, int) {
	s.mu.RLock()
	defer s.mu.RUnlock()
	e, ok := s.m[string(key)]
	if !ok || e.expired(now) {
		return 0, -2
	}
	if e.expires.IsZero() {
		return 0, -1
	}
	return e.expires.Sub(now), 0
}

// Sweep removes keys past their deadline, looking at no more than budget of them. The
// map iterates in a different order each time, so each call sees a fresh sample.
func (s *Store) Sweep(now time.Time, budget int) int {
	s.mu.Lock()
	defer s.mu.Unlock()
	removed := 0
	for k, e := range s.m {
		if budget == 0 {
			break
		}
		budget--
		if e.expired(now) {
			delete(s.m, k)
			removed++
		}
	}
	return removed
}

/// A value and its deadline. `None` means the key never expires. An `Instant` is a
/// reading of the monotonic clock, so a change to the system time cannot move it.
struct Entry {
    val: Value,
    expires: Option<Instant>,
}

impl Entry {
    fn expired(&self, now: Instant) -> bool {
        self.expires.is_some_and(|at| at <= now)
    }
}

fn set_expiry(&mut self, key: &[u8], at: Instant, now: Instant) -> bool {
    match self.map.get_mut(key) {
        Some(e) if !e.expired(now) => {
            e.expires = Some(at);
            true
        }
        _ => false,
    }
}

fn ttl(&self, key: &[u8], now: Instant) -> Ttl {
    match self.map.get(key) {
        None => Ttl::Missing,
        Some(e) if e.expired(now) => Ttl::Missing,
        Some(Entry { expires: None, .. }) => Ttl::Never,
        Some(Entry {
            expires: Some(at), ..
        }) => Ttl::Left(*at - now),
    }
}

/// `retain` walks the whole map, which is fine for a store of this size; Redis
/// samples instead, to bound the time.
fn sweep(&mut self, now: Instant) -> usize {
    let before = self.map.len();
    self.map.retain(|_, e| !e.expired(now));
    before - self.map.len()
}

/// SET key value [EX seconds | PX milliseconds]. A plain SET removes any deadline, as
/// Redis does; a SET with EX or PX sets a new one.
fn set(args: &Command, store: &mut dyn Store, now: Instant, w: &mut dyn Write) -> io::Result<()> {
    match args.as_slice() {
        [_, key, val] => {
            // the one copy: the store must own its key and value, the buffer does not
            store.set(key.to_vec(), val.to_vec());
            resp::write_simple(w, "OK")
        }
        [_, key, val, unit, n] => {
            let ttl = match (unit.to_ascii_uppercase().as_slice(), parse_int(n)) {
                (b"EX" | b"PX", None) => {
                    return resp::write_error(w, "ERR value is not an integer or out of range");
                }
                (b"EX", Some(n)) if n > 0 && n <= i64::MAX / 1000 => Duration::from_secs(n as u64),
                (b"PX", Some(n)) if n > 0 => Duration::from_millis(n as u64),
                (b"EX" | b"PX", Some(_)) => {
                    return resp::write_error(w, "ERR invalid expire time in 'set' command");
                }
                _ => return resp::write_error(w, "ERR syntax error"),
            };
            let Some(at) = now.checked_add(ttl) else {
                return resp::write_error(w, "ERR invalid expire time in 'set' command");
            };
            store.set(key.to_vec(), val.to_vec());
            store.set_expiry(key, at, now);
            resp::write_simple(w, "OK")
        }
        _ => write_arity(w, b"set"),
    }
}

/* A write to a client that has gone raises SIGPIPE, which kills the process by
 * default. Ignore it, and write() reports EPIPE instead, which closes that client.
 * Linux can ask for the same with MSG_NOSIGNAL on send(), and macOS with the
 * SO_NOSIGPIPE socket option. Ignoring the signal covers both. */
signal(SIGPIPE, SIG_IGN);
/* Ctrl-C and kill send SIGINT and SIGTERM, which also end the process by default.
 * From here on they end the event loop instead, through a pipe it waits on. */
int stop_fd = kv_stop_fd();
if (stop_fd < 0) {
    perror("kv_stop_fd");
    return 1;
}

/* Which signal asked the server to stop, or 0. sig_atomic_t is the one integer type
 * that a signal handler may write while the program reads it, without a torn value, and
 * volatile tells the compiler the value can change behind its back. */
extern volatile sig_atomic_t kv_stop_signal;

/* A signal handler runs between any two instructions of the program, even in the middle
 * of malloc or printf. So it may only do what is safe at any moment: set the flag, and
 * call functions on the async-signal-safe list. write() is on the list. The byte wakes
 * the loop, which is otherwise asleep in its wait and would not look at the flag. */
static void kv_on_stop(int sig) {
    int saved = errno; /* write() may change errno, and the interrupted code needs it */
    kv_stop_signal = sig;
    if (write(kv_stop_write, "s", 1) < 0) {
        /* the pipe is full or closed: the loop is already awake, or already gone */
    }
    errno = saved;
}

int kv_stop_fd(void) {
    int fds[2];
    if (pipe(fds) < 0) {
        return -1;
    }
    /* non-blocking, so that a handler can never wait on a full pipe */
    for (int i = 0; i < 2; i++) {
        int flags = fcntl(fds[i], F_GETFL, 0);
        if (flags < 0 || fcntl(fds[i], F_SETFL, flags | O_NONBLOCK) < 0) {
            return -1;
        }
    }
    kv_stop_write = fds[1];
    /* sigaction, not signal(): it says exactly what happens, on every Unix. SA_RESTART
     * lets a read or accept that the signal interrupted carry on, because the pipe,
     * not the interruption, is what tells the loop to stop. */
    struct sigaction sa = {0};
    sa.sa_handler = kv_on_stop;
    sigemptyset(&sa.sa_mask);
    sa.sa_flags = SA_RESTART;
    if (sigaction(SIGINT, &sa, NULL) < 0 || sigaction(SIGTERM, &sa, NULL) < 0) {
        return -1;
    }
    return fds[0];
}

/* A byte on the pipe is the signal, delivered to the loop as an event like
 * any other. Nothing was interrupted in the middle, so there is nothing to
 * repair: the loop ends, and main closes everything in order. */
if (c == (kv_conn *)&stop_tag) {
    rc = 0;
    break;
}

// Ctrl-C and kill send SIGINT and SIGTERM, which end a Go program at once by default.
// NotifyContext changes that: the signal cancels a context instead, and every part of
// the server that holds the context sees it. A second signal, after stop() has run,
// ends the program the default way, for a server that will not finish.
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop()

// Three parts to a clean stop. Closing the listener makes Accept return an error,
// which ends the loop. AfterFunc closes each client when the context is cancelled,
// which ends its serve. The WaitGroup waits for every serve to return.
var clients sync.WaitGroup
context.AfterFunc(ctx, func() { ln.Close() })
for {
	conn, err := ln.Accept()
	if err != nil {
		if ctx.Err() != nil {
			break
		}
		fmt.Fprintln(os.Stderr, "accept:", err)
		continue
	}
	clients.Go(func() {
		defer context.AfterFunc(ctx, func() { conn.Close() })()
		if err := serve(conn, s); err != nil && ctx.Err() == nil {
			fmt.Fprintln(os.Stderr, "client:", err)
		}
	})
}
fmt.Fprintln(os.Stderr, "kv stopping:", context.Cause(ctx))
clients.Wait()

// A panic in one command would end the whole process, and every other client with
// it. recover, which only works inside a deferred function, turns it into an error
// for this one connection, which closes. The bug is still logged, and still a bug.
defer func() {
	if r := recover(); r != nil {
		err = fmt.Errorf("panic: %v", r)
	}
}()

/// Block SIGINT and SIGTERM in the calling thread. Every thread spawned afterwards
/// inherits the mask, so the one thread that calls `wait_for_stop` is the only one that
/// ever sees them, and no handler runs in the middle of anything.
#[cfg(unix)]
pub fn block_stop_signals() {
    // SAFETY: the set is initialised by sigemptyset before any use, and the mask call
    // reads it and writes nothing else.
    unsafe {
        let mut set = std::mem::zeroed::<libc::sigset_t>();
        libc::sigemptyset(&mut set);
        libc::sigaddset(&mut set, libc::SIGINT);
        libc::sigaddset(&mut set, libc::SIGTERM);
        libc::pthread_sigmask(libc::SIG_BLOCK, &set, std::ptr::null_mut());
    }
}

/// Wait until SIGINT or SIGTERM arrives, and say which. sigwait takes the signal off
/// the pending list and returns it to an ordinary thread, which may then do anything.
#[cfg(unix)]
pub fn wait_for_stop() -> i32 {
    // SAFETY: as above, and sigwait writes one int through a valid pointer.
    unsafe {
        let mut set = std::mem::zeroed::<libc::sigset_t>();
        libc::sigemptyset(&mut set);
        libc::sigaddset(&mut set, libc::SIGINT);
        libc::sigaddset(&mut set, libc::SIGTERM);
        let mut sig = 0;
        libc::sigwait(&set, &mut sig);
        sig
    }
}

/// Elsewhere there is nothing to block, and nothing to wait for: Ctrl-C ends the
/// process the default way.
#[cfg(not(unix))]
pub fn block_stop_signals() {}

#[cfg(not(unix))]
pub fn wait_for_stop() -> i32 {
    loop {
        std::thread::park();
    }
}

// One thread waits for the signal. It raises the flag and then wakes the accept
// loop the only way a plain listener can be woken: by connecting to it.
s.spawn(|| {
    let sig = sys::wait_for_stop();
    eprintln!("kv stopping on signal {sig}");
    stopping.store(true, Ordering::Relaxed);
    let _ = TcpStream::connect(("127.0.0.1", port.parse().unwrap_or(6380)));
});

// Shutting a socket makes its read return 0, and that client's serve returns.
// The scope then joins every thread, so nothing is still running when main ends.
for c in clients.lock().unwrap_or_else(|e| e.into_inner()).iter() {
    let _ = c.shutdown(Shutdown::Both);
}

/// Completes when Ctrl-C or kill asks the process to stop. Asking tokio for the signal
/// is what turns off the default, which would end the process at once.
async fn stop_signal() {
    #[cfg(unix)]
    {
        use tokio::signal::unix::{SignalKind, signal};
        let mut term = signal(SignalKind::terminate()).expect("SIGTERM handler");
        tokio::select! {
            _ = tokio::signal::ctrl_c() => {}
            _ = term.recv() => {}
        }
    }
    #[cfg(not(unix))]
    {
        let _ = tokio::signal::ctrl_c().await;
    }
}

// Two channels for a clean stop. The broadcast channel tells: every task holds a
// receiver, and one send reaches all of them. The mpsc channel waits: every task
// holds a sender it never uses, and recv() returns None once the last one is dropped.
let (stop, _) = broadcast::channel::<()>(1);
let (done, mut all_done) = mpsc::channel::<()>(1);

loop {
    let stream = tokio::select! {
        accepted = listener.accept() => accepted?.0,
        _ = stop_signal() => break,
    };
    let store = Arc::clone(&store);
    let mut stop_rx = stop.subscribe();
    let done = done.clone();
    tokio::spawn(async move {
        // A task waits on its client and on the stop at once. When the stop wins,
        // the serve future is dropped where it stands, between two awaits, and that
        // is the whole of cancellation: nothing to flag, nothing to catch.
        tokio::select! {
            res = serve(stream, store) => {
                if let Err(e) = res {
                    eprintln!("client: {e}");
                }
            }
            _ = stop_rx.recv() => {}
        }
        drop(done);
    });
}
eprintln!("kv (tokio) stopping");
let _ = stop.send(());
drop(done);
all_done.recv().await; // None, once every task has dropped its sender

/// A parsed request: the command name and its arguments, as slices of the buffer the
/// request arrived in. Nothing is copied. The lifetime `'a` is the buffer's, and it says
/// that a `Command` cannot outlive the bytes it points into. A command that needs to
/// keep an argument, as SET keeps its value, copies it then, and only then.
pub type Command<'a> = Vec<&'a [u8]>;

/// What `parse` found in the bytes so far.
#[derive(Debug)]
pub enum Parsed<'a> {
    /// A whole request, and how many bytes of the buffer it used.
    Done(Command<'a>, usize),
    /// Not a whole request yet: read more bytes and call again.
    Incomplete,
}

/// One request from the front of `buf`: an array of bulk strings such as
/// `*2\r\n$4\r\nECHO\r\n$2\r\nhi\r\n`, or one inline line of words. The returned
/// `Command` borrows `buf`, which the signature says with the elided lifetime: one
/// reference in, so the one reference out lives as long as it.
pub fn parse(buf: &[u8]) -> Result<Parsed<'_>> {
    let Some((line, mut used)) = take_line(buf, 0) else {
        return Ok(Parsed::Incomplete);
    };
    if line.first() != Some(&b'*') {
        let words = line.split(|b| *b == b' ').filter(|w| !w.is_empty());
        return Ok(Parsed::Done(words.collect(), used));
    }
    let n = parse_len(&line[1..], b'*')?;
    if n > MAX_ARGS {
        return Err(KvError::Protocol(b'*'));
    }
    let mut args = Vec::with_capacity(n);
    for _ in 0..n {
        let Some((line, start)) = take_line(buf, used) else {
            return Ok(Parsed::Incomplete);
        };
        if line.first() != Some(&b'$') {
            return Err(KvError::Protocol(line.first().copied().unwrap_or(b'\n')));
        }
        let size = parse_len(&line[1..], b'$')?;
        if size > MAX_BULK {
            return Err(KvError::Protocol(b'$'));
        }
        let end = start + size;
        if buf.len() < end + 2 {
            return Ok(Parsed::Incomplete);
        }
        if &buf[end..end + 2] != b"\r\n" {
            return Err(KvError::Protocol(buf[end]));
        }
        args.push(&buf[start..end]);
        used = end + 2;
    }
    Ok(Parsed::Done(args, used))
}

/// The line that starts at `from`, without its ending, and where the next line starts.
/// A bare LF is accepted as well as CRLF, so that a command typed into nc works.
fn take_line(buf: &[u8], from: usize) -> Option<(&[u8], usize)> {
    let rest = &buf[from..];
    let nl = rest.iter().position(|b| *b == b'\n')?;
    let line = rest[..nl].strip_suffix(b"\r").unwrap_or(&rest[..nl]);
    Some((line, from + nl + 1))
}

fn parse_len(digits: &[u8], type_byte: u8) -> Result<usize> {
    std::str::from_utf8(digits)
        .ok()
        .and_then(|s| s.parse().ok())
        .ok_or(KvError::Protocol(type_byte))
}

/// Bytes are read into a buffer and whole requests are parsed out of it without a
/// copy. The store is shared: the Mutex lets one thread at a time use it, and the scope
/// in main lends it to every client, so no Arc is needed.
fn serve(mut stream: TcpStream, store: &Mutex<MapStore>) -> resp::Result<()> {
    let mut writer = BufWriter::new(stream.try_clone()?);
    let mut buf: Vec<u8> = Vec::with_capacity(4096);
    let mut chunk = [0u8; 4096];
    loop {
        let n = stream.read(&mut chunk)?;
        if n == 0 {
            return Ok(()); // the client closed, or the server shut the socket
        }
        buf.extend_from_slice(&chunk[..n]);
        let mut used = 0;
        loop {
            match resp::parse(&buf[used..]) {
                Ok(Parsed::Done(args, n)) => {
                    let mut store = store.lock().unwrap_or_else(|e| e.into_inner());
                    cmd::execute(&args, &mut *store, &mut writer)?;
                    used += n;
                }
                Ok(Parsed::Incomplete) => break,
                Err(e @ resp::KvError::Protocol(_)) => {
                    resp::write_error(&mut writer, "ERR Protocol error")?;
                    writer.flush()?;
                    return Err(e);
                }
                Err(e) => return Err(e),
            }
        }
        // the arguments borrowed the buffer, and they are gone, so the buffer may change
        buf.drain(..used);
        if buf.len() > resp::MAX_REQUEST {
            return Err(resp::KvError::Protocol(b'*')); // a request that never ends
        }
        writer.flush()?;
    }
}

// buf[:size] would share its spare two bytes with the line ending. The full slice
// expression caps the capacity at the length, so an append to the argument copies.
args = append(args, buf[:size:size])

// Every argument is a slice into a buffer the parser allocated, with two bytes of line
// ending after it. The full slice expression buf[:size:size] sets the capacity to the
// length, so that an append to an argument must copy instead of writing over the line
// ending, or over whatever a later version of the parser keeps there.
func TestArgumentsDoNotShareSpareCapacity(t *testing.T) {
	r := bufio.NewReader(strings.NewReader("*3\r\n$3\r\nSET\r\n$1\r\nk\r\n$2\r\nhi\r\n"))
	args, err := ReadCommand(r)
	if err != nil {
		t.Fatal(err)
	}
	val := args[2]
	if cap(val) != len(val) {
		t.Fatalf("cap %d, len %d: an append would write into the parser's buffer", cap(val), len(val))
	}
	grown := append(val, '!')
	if &grown[0] == &val[0] {
		t.Fatal("append reused the argument's memory")
	}
	if string(val) != "hi" || string(grown) != "hi!" {
		t.Fatalf("val %q, grown %q", val, grown)
	}
}

/* One parsed request. The arguments point into the request buffer, so they are valid
 * only until that buffer is consumed. A kv_cmd owns nothing: the comment is the whole
 * of the rule, and const is the part of it the compiler can check. */
typedef struct {
    size_t argc;
    const char *argv[KV_MAX_ARGS];
    size_t arglen[KV_MAX_ARGS];
} kv_cmd;

/* How often the file is forced to disk. Redis has the same three settings. */
typedef enum {
    KV_FSYNC_ALWAYS,   /* after every command: slow and safe */
    KV_FSYNC_EVERYSEC, /* once a second: lose at most a second of writes */
    KV_FSYNC_NO        /* never: the operating system decides */
} kv_fsync_policy;

typedef struct {
    int fd; /* -1 when persistence is off */
    kv_fsync_policy policy;
    bool dirty;           /* bytes written since the last fsync was asked for */
    int64_t last_request; /* monotonic ms of the last request to the syncer */
    kv_syncer syncer;     /* the thread that waits for the disk */
} kv_aof;

bool kv_aof_open(kv_aof *a, const char *path, kv_fsync_policy policy) {
    /* O_APPEND: every write goes to the end, even if another process appends too */
    int fd = open(path, O_WRONLY | O_CREAT | O_APPEND, 0644);
    if (fd < 0) {
        perror(path);
        return false;
    }
    *a = (kv_aof){.fd = fd, .policy = policy, .last_request = kv_now_ms()};
    if (!kv_syncer_start(&a->syncer, fd)) {
        perror("kv_syncer_start");
        close(fd);
        a->fd = -1;
        return false;
    }
    return true;
}

static bool kv_write_all(int fd, const char *p, size_t n) {
    size_t off = 0;
    while (off < n) {
        ssize_t w = write(fd, p + off, n - off);
        if (w < 0) {
            if (errno == EINTR) {
                continue;
            }
            perror("aof write");
            return false;
        }
        off += (size_t)w;
    }
    return true;
}

/* PEXPIREAT key <ms since 1970>, as the absolute form of EX, PX and EXPIRE. */
static bool kv_encode_pexpireat(kv_buf *out, const char *key, size_t klen, int64_t at) {
    char digits[32];
    int n = snprintf(digits, sizeof digits, "%lld", (long long)at);
    const char *argv[3] = {"PEXPIREAT", key, digits};
    size_t arglen[3] = {9, klen, (size_t)n};
    return kv_aof_encode(out, 3, argv, arglen);
}

bool kv_aof_record(kv_aof *a, const kv_cmd *c, int64_t wall_now) {
    if (a->fd < 0) {
        return true;
    }
    kv_buf out = {0};
    bool ok;
    long long n;
    if (kv_arg_is(c, 0, "SET") && c->argc == 5 && kv_parse_int(c->argv[4], c->arglen[4], &n) &&
        n > 0) {
        /* SET key value EX|PX n becomes SET key value, then PEXPIREAT key <absolute> */
        int64_t at = wall_now + (kv_arg_is(c, 3, "EX") ? n * 1000 : n);
        ok = kv_aof_encode(&out, 3, c->argv, c->arglen) &&
             kv_encode_pexpireat(&out, c->argv[1], c->arglen[1], at);
    } else if (kv_arg_is(c, 0, "EXPIRE") && c->argc == 3 &&
               kv_parse_int(c->argv[2], c->arglen[2], &n)) {
        if (n <= 0) {
            const char *argv[2] = {"DEL", c->argv[1]};
            size_t arglen[2] = {3, c->arglen[1]};
            ok = kv_aof_encode(&out, 2, argv, arglen);
        } else {
            ok = kv_encode_pexpireat(&out, c->argv[1], c->arglen[1], wall_now + n * 1000);
        }
    } else {
        ok = kv_aof_encode(&out, c->argc, c->argv, c->arglen);
    }
    ok = ok && kv_write_all(a->fd, out.data, out.len);
    kv_buf_free(&out);
    if (!ok) {
        return false;
    }
    a->dirty = true;
    if (a->policy == KV_FSYNC_ALWAYS) {
        /* always: wait for the disk here, before the reply, on the one thread */
        if (!kv_fsync(a->fd)) {
            perror("fsync");
        }
        a->dirty = false;
    }
    return true;
}

/* The command ran; now it goes to the file, before the reply goes to the client. */
if (kv_cmd_is_write(&cmd) && !kv_aof_record(aof, &cmd, kv_wall_ms())) {
    return false;
}

typedef struct {
    pthread_t thread;
    pthread_mutex_t mu; /* guards wanted and stop */
    pthread_cond_t cv;  /* signalled when either changes */
    bool wanted;        /* an fsync has been asked for */
    bool stop;          /* the thread should return */
    int fd;
    /* when the last fsync finished, in monotonic ms; written by the thread and read by
     * the loop without the lock, which is what an atomic is for */
    _Atomic int64_t last_sync;
} kv_syncer;

/* The thread. It sleeps on the condition variable until the loop asks for an fsync or
 * for it to stop, and it never holds the mutex while it waits for the disk. */
static void *kv_syncer_run(void *arg) {
    kv_syncer *s = arg;
    pthread_mutex_lock(&s->mu);
    for (;;) {
        /* a while, not an if: a wait may return early, and the loop may change its mind */
        while (!s->wanted && !s->stop) {
            pthread_cond_wait(&s->cv, &s->mu); /* releases mu while asleep, takes it back after */
        }
        if (s->stop) {
            break;
        }
        s->wanted = false;
        pthread_mutex_unlock(&s->mu);
        if (!kv_fsync(s->fd)) {
            perror("fsync");
        }
        /* release: every write the fsync made visible is published with the time */
        atomic_store_explicit(&s->last_sync, kv_now_ms(), memory_order_release);
        pthread_mutex_lock(&s->mu);
    }
    pthread_mutex_unlock(&s->mu);
    return NULL;
}

bool kv_syncer_start(kv_syncer *s, int fd) {
    *s = (kv_syncer){.fd = fd};
    atomic_init(&s->last_sync, kv_now_ms());
    if (pthread_mutex_init(&s->mu, NULL) != 0 || pthread_cond_init(&s->cv, NULL) != 0) {
        return false;
    }
    /* A signal is delivered to any thread that does not block it. The stop handler must
     * run on the event loop's thread, so this thread is created with every signal
     * blocked, which it inherits, and the loop's mask is put back afterwards. */
    sigset_t all, before;
    sigfillset(&all);
    pthread_sigmask(SIG_BLOCK, &all, &before);
    int rc = pthread_create(&s->thread, NULL, kv_syncer_run, s);
    pthread_sigmask(SIG_SETMASK, &before, NULL);
    return rc == 0;
}

void kv_syncer_request(kv_syncer *s) {
    pthread_mutex_lock(&s->mu);
    s->wanted = true;
    pthread_cond_signal(&s->cv);
    pthread_mutex_unlock(&s->mu);
}

int64_t kv_syncer_last(const kv_syncer *s) {
    /* acquire: pairs with the release in the thread, so the reader sees the fsync */
    return atomic_load_explicit(&s->last_sync, memory_order_acquire);
}

void kv_syncer_stop(kv_syncer *s) {
    pthread_mutex_lock(&s->mu);
    s->stop = true;
    pthread_cond_signal(&s->cv);
    pthread_mutex_unlock(&s->mu);
    pthread_join(s->thread, NULL);
    pthread_cond_destroy(&s->cv);
    pthread_mutex_destroy(&s->mu);
}

void kv_aof_tick(kv_aof *a, int64_t now) {
    if (a->fd < 0 || !a->dirty || a->policy != KV_FSYNC_EVERYSEC) {
        return;
    }
    if (now - a->last_request < 1000) {
        return; /* not yet a second since the last request */
    }
    if (kv_syncer_last(&a->syncer) < a->last_request) {
        return; /* the last fsync is still running: a slow disk gets one at a time */
    }
    a->dirty = false;
    a->last_request = now;
    kv_syncer_request(&a->syncer); /* returns at once; the thread waits for the disk */
}

bool kv_fsync(int fd) {
#ifdef __APPLE__
    /* On macOS fsync() only hands the bytes to the drive, which may hold them in its own
     * cache. F_FULLFSYNC asks the drive to write them, and is what Go and Rust call. */
    return fcntl(fd, F_FULLFSYNC) >= 0;
#else
    return fsync(fd) >= 0;
#endif
}

/* PEXPIREAT key <ms since 1970>: the absolute form that the append-only file uses. The
 * wall-clock deadline is turned into a monotonic one for the store. */
static bool kv_cmd_pexpireat(const kv_cmd *c, kv_store *s, kv_buf *out, int64_t now) {
    long long at;
    if (!kv_parse_int(c->argv[2], c->arglen[2], &at)) {
        return kv_reply_error(out, "ERR value is not an integer or out of range");
    }
    kv_entry *e = kv_store_find(s, c->argv[1], c->arglen[1], now);
    if (e == NULL) {
        return kv_reply_int(out, 0);
    }
    int64_t left = at - kv_wall_ms();
    if (left <= 0) {
        kv_store_del(s, c->argv[1], c->arglen[1]);
    } else {
        e->expires_at = now + left;
    }
    return kv_reply_int(out, 1);
}

/* KV_AOF names the file. Replay it first, then open it for appending. */
kv_aof aof = {.fd = -1, .policy = KV_FSYNC_EVERYSEC};
const char *path = getenv("KV_AOF");
if (path != NULL && *path != '\0') {
    kv_fsync_policy policy;
    if (!kv_fsync_policy_parse(getenv("KV_FSYNC"), &policy)) {
        fprintf(stderr, "KV_FSYNC must be always, everysec or no\n");
        return 1;
    }
    long applied = kv_aof_load(path, store);
    if (applied < 0) {
        perror(path);
        return 1;
    }
    fprintf(stderr, "kv replayed %ld commands from %s\n", applied, path);
    if (!kv_aof_open(&aof, path, policy)) {
        return 1;
    }
}

// Policy says how often the file is forced to disk. Redis has the same three settings.
type Policy int

const (
	Always   Policy = iota // after every command: slow and safe
	EverySec               // once a second: lose at most a second of writes
	No                     // never: the operating system decides
)

func ParsePolicy(s string) (Policy, error) {
	switch strings.ToLower(s) {
	case "", "everysec":
		return EverySec, nil
	case "always":
		return Always, nil
	case "no":
		return No, nil
	}
	return 0, fmt.Errorf("KV_FSYNC must be always, everysec or no, not %q", s)
}

// Log is the open file. Many goroutines record into it, so it has a lock of its own.
type Log struct {
	mu     sync.Mutex
	f      *os.File
	w      *bufio.Writer
	policy Policy
	dirty  bool
	syncs chan struct{} // a request to the syncer goroutine; room for one, so a second waits for nobody
	done  sync.WaitGroup
}

// Record appends one command. A relative expiry is rewritten as an absolute PEXPIREAT,
// because the file may be replayed much later. The bytes reach the kernel before this
// returns: a policy delays only the fsync, never the write, so a crash of the process
// loses nothing and only a crash of the machine can lose the last second.
func (l *Log) Record(args [][]byte, now time.Time) error {
	l.mu.Lock()
	defer l.mu.Unlock()
	for _, cmd := range rewrite(args, now) {
		if err := resp.WriteArrayLen(l.w, len(cmd)); err != nil {
			return err
		}
		for _, a := range cmd {
			if err := resp.WriteBulk(l.w, a); err != nil {
				return err
			}
		}
	}
	if err := l.w.Flush(); err != nil {
		return err
	}
	l.dirty = true
	if l.policy == Always {
		return l.syncLocked()
	}
	return nil
}

// rewrite turns one command into the commands the file stores.
func rewrite(args [][]byte, now time.Time) [][][]byte {
	name := strings.ToUpper(string(args[0]))
	switch {
	case name == "SET" && len(args) == 5:
		n, err := strconv.ParseInt(string(args[4]), 10, 64)
		if err != nil || n <= 0 {
			break
		}
		d := time.Duration(n) * time.Millisecond
		if strings.ToUpper(string(args[3])) == "EX" {
			d = time.Duration(n) * time.Second
		}
		return [][][]byte{args[:3], pexpireat(args[1], now.Add(d))}
	case name == "EXPIRE" && len(args) == 3:
		n, err := strconv.ParseInt(string(args[2]), 10, 64)
		if err != nil {
			break
		}
		if n <= 0 {
			return [][][]byte{{[]byte("DEL"), args[1]}}
		}
		return [][][]byte{pexpireat(args[1], now.Add(time.Duration(n)*time.Second))}
	}
	return [][][]byte{args}
}

func pexpireat(key []byte, at time.Time) [][]byte {
	return [][]byte{[]byte("PEXPIREAT"), key, []byte(strconv.FormatInt(at.UnixMilli(), 10))}
}

// syncer runs on a goroutine of its own and calls fsync whenever it is asked, so that no
// client ever waits for the disk while holding the log's lock. The range ends when Close
// closes the channel, and the WaitGroup lets Close wait for that.
func (l *Log) syncer() {
	for range l.syncs {
		if err := l.f.Sync(); err != nil {
			fmt.Fprintln(os.Stderr, "aof:", err)
		}
	}
}

// Tick is called once a second. With everysec it flushes the buffer to the kernel and
// asks the syncer for an fsync, without waiting for it. The select with a default never
// blocks: if a request is already waiting in the channel, this one is not needed.
func (l *Log) Tick() {
	l.mu.Lock()
	defer l.mu.Unlock()
	if !l.dirty || l.policy != EverySec {
		return
	}
	if err := l.w.Flush(); err != nil {
		fmt.Fprintln(os.Stderr, "aof:", err)
		return
	}
	l.dirty = false
	select {
	case l.syncs <- struct{}{}:
	default:
	}
}

// Execute runs one command against the store and writes its reply. A write command is
// also recorded in the log, when there is one. During replay the log is nil.
func Execute(args [][]byte, s Storer, log *aof.Log, w *bufio.Writer) error {
	now := time.Now()
	if len(args) == 0 {
		return resp.WriteError(w, "ERR empty command")
	}
	name := strings.ToUpper(string(args[0]))
	cmd, ok := commands[name]
	if !ok {
		shown := string(args[0])
		if len(shown) > 40 {
			shown = shown[:40]
		}
		return resp.WriteError(w, "ERR unknown command '"+shown+"'")
	}
	if n := len(args); (cmd.arity < 0 && n < -cmd.arity) || (cmd.arity > 0 && n != cmd.arity) {
		return arity(w, strings.ToLower(name))
	}
	if log != nil && aof.IsWrite(name) {
		if err := log.Record(args, now); err != nil {
			return err
		}
	}
	return cmd.run(args, s, w, now)
}

// KV_AOF names the file. Replay it first, with no log, then open it for appending.
var log *aof.Log
if path := os.Getenv("KV_AOF"); path != "" {
	policy, err := aof.ParsePolicy(os.Getenv("KV_FSYNC"))
	if err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(1)
	}
	sink := bufio.NewWriter(io.Discard)
	n, err := aof.Replay(path, func(args [][]byte) { cmd.Execute(args, s, nil, sink) })
	if err != nil {
		fmt.Fprintln(os.Stderr, "aof:", err)
		os.Exit(1)
	}
	fmt.Fprintf(os.Stderr, "kv replayed %d commands from %s\n", n, path)
	if log, err = aof.Open(path, policy); err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(1)
	}
	go func() {
		tick := time.NewTicker(time.Second)
		defer tick.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-tick.C:
				log.Tick()
			}
		}
	}()
}

/// Everything a command may touch. One Mutex around this struct means a command and its
/// record in the log happen together, with no other command in between.
pub struct Db {
    pub store: MapStore,
    pub aof: Option<Aof>,
}

/// Append one command. A relative expiry is rewritten as an absolute PEXPIREAT,
/// because the file may be replayed much later. The bytes reach the kernel before this
/// returns: a policy delays only the fsync, never the write, so a crash of the process
/// loses nothing and only a crash of the machine can lose the last second.
pub fn record(&mut self, args: &Command<'_>, now_ms: i64) -> io::Result<()> {
    for cmd in rewrite(args, now_ms) {
        resp::write_array_len(&mut self.file, cmd.len())?;
        for a in &cmd {
            resp::write_bulk(&mut self.file, a)?;
        }
    }
    self.file.flush()?;
    self.dirty = true;
    if self.policy == Policy::Always {
        self.sync()?;
    }
    Ok(())
}

/// Flush the buffer and force the file to disk, on the calling thread. The always
/// policy does this after every record, and main does it once, last of all.
pub fn sync(&mut self) -> io::Result<()> {
    self.file.flush()?;
    if mem::replace(&mut self.dirty, false) && self.policy != Policy::No {
        self.file.get_ref().sync_all()?;
    }
    Ok(())
}

impl Aof {
    /// A second handle to the same file, for the thread that waits for the disk. An
    /// fsync through either handle forces everything written through both.
    pub fn handle(&self) -> io::Result<File> {
        self.file.get_ref().try_clone()
    }

    /// Once a second, with everysec: flush the buffer to the kernel and say whether an
    /// fsync is owed. `mem::replace` reads the flag and clears it in one expression.
    pub fn owes_sync(&mut self) -> io::Result<bool> {
        self.file.flush()?;
        Ok(self.policy == Policy::EverySec && mem::replace(&mut self.dirty, false))
    }
}

// The thread that waits for the disk. It receives a request on the channel,
// calls fsync through its own handle, and never touches the lock. The loop ends
// when every sender is dropped, which happens when the sweeper returns.
let (sync_tx, sync_rx) = mpsc::channel::<()>();
if let Some(file) = sync_file {
    s.spawn(move || {
        for () in sync_rx {
            if let Err(e) = file.sync_all() {
                eprintln!("aof: {e}");
            }
        }
    });
}

// every tenth tick, once a second, the everysec policy flushes the log and
// asks the syncer for an fsync, without waiting for it
if tick.is_multiple_of(10) {
    match db.aof.as_mut().map(Aof::owes_sync) {
        Some(Ok(true)) => {
            let _ = sync_tx.send(());
        }
        Some(Err(e)) => eprintln!("aof: {e}"),
        _ => {}
    }
}

// fsync blocks, and a task must not. spawn_blocking runs the call on
// a thread kept for exactly that, and the task goes on at once.
if owed && let Some(file) = sync_file.clone() {
    tokio::task::spawn_blocking(move || {
        if let Err(e) = file.sync_all() {
            eprintln!("aof: {e}");
        }
    });
}

/// Replay the file into the store, through the same parser a client goes through, from
/// the same kind of buffer. A truncated last command is dropped, as Redis drops it. A
/// missing file is an empty store.
pub fn replay(path: &Path, store: &mut MapStore) -> io::Result<usize> {
    let mut file = match File::open(path) {
        Ok(f) => f,
        Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(0),
        Err(e) => return Err(e),
    };
    let mut sink = io::sink();
    let mut buf: Vec<u8> = Vec::new();
    let mut chunk = vec![0u8; 1 << 16];
    let mut n = 0;
    loop {
        let got = file.read(&mut chunk)?;
        if got == 0 {
            if !buf.is_empty() {
                eprintln!("aof: dropped a truncated command at the end");
            }
            return Ok(n);
        }
        buf.extend_from_slice(&chunk[..got]);
        let mut used = 0;
        while let Parsed::Done(args, len) = resp::parse(&buf[used..])? {
            crate::cmd::execute(&args, store, &mut sink)?;
            n += 1;
            used += len;
        }
        buf.drain(..used);
    }
}

static int failed = 0;
static int passed = 0;

/* Report a failure with its location, and carry on to the next check. */
#define CHECK(cond)                                                                                \
    do {                                                                                           \
        if (cond) {                                                                                \
            passed++;                                                                              \
        } else {                                                                                   \
            failed++;                                                                              \
            fprintf(stderr, "%s:%d: check failed: %s\n", __FILE__, __LINE__, #cond);               \
        }                                                                                          \
    } while (0)

/* The tests assume a request may carry at least four arguments. The compiler checks it,
 * so a smaller limit fails to build instead of failing at run time. */
_Static_assert(KV_MAX_ARGS >= 4, "the tests use requests of up to four arguments");

/* The list of tests, written once. Each use of the list defines X to mean something
 * different: a declaration, an entry in a table, or a call. Adding a test is one line
 * here, and nothing else to keep in step. */
#define KV_TESTS                                                                                   \
    X(parse_complete)                                                                              \
    X(parse_incomplete_at_every_cut)                                                               \
    X(parse_bad)                                                                                   \
    X(parse_inline)                                                                                \
    X(parse_int)                                                                                   \
    X(glob)

static const struct {
    const char *name;
    void (*run)(void);
} tests[] = {
#define X(name) {#name, test_##name},
    KV_TESTS
#undef X
};

/* Run every test, or only those whose name contains the argument: ./test_resp parse */
int main(int argc, char **argv) {
    const char *only = argc > 1 ? argv[1] : "";
    int ran = 0;
    for (size_t i = 0; i < sizeof tests / sizeof tests[0]; i++) {
        if (strstr(tests[i].name, only) == NULL) {
            continue;
        }
        int before = failed;
        tests[i].run();
        printf("%s %s\n", failed == before ? "ok  " : "FAIL", tests[i].name);
        ran++;
    }
    printf("%d tests, %d checks passed, %d failed\n", ran, passed, failed);
    return failed == 0 ? 0 : 1;
}

# The unit tests: every source under src/ except main.c, plus the test file, compiled
# together under the sanitizers. make test fails when a check fails.

test: build/test_resp
	./build/test_resp

build/test_resp: tests/test_resp.c $(SRC) | build
	$(CC) $(CFLAGS) -fsanitize=address,undefined -o $@ tests/test_resp.c $(filter-out src/main.c,$(SRC))

int main(void) {
    kv_buf b = {0};
    kv_buf_append(&b, "GET key\r\n", 9);
    const char *key = b.data + 4; /* points at "key", inside the buffer */
    char filler[2048];
    memset(filler, 'x', sizeof filler);
    kv_buf_append(&b, filler, sizeof filler); /* the buffer grows, and may move */
    printf("%.3s\n", key);                    /* reads through the old pointer */
    kv_buf_free(&b);
    return 0;
}

static long counter = 0;

static void *add(void *arg) {
    (void)arg;
    for (int i = 0; i < 100000; i++) {
        counter++; /* a read, an add and a write, and another thread between them */
    }
    return NULL;
}

int main(void) {
    pthread_t a, b;
    pthread_create(&a, NULL, add, NULL);
    pthread_create(&b, NULL, add, NULL);
    pthread_join(a, NULL);
    pthread_join(b, NULL);
    printf("%ld\n", counter); /* 200000 if nothing went wrong, which something did */
    return 0;
}

int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) {
    kv_store store = {0};
    kv_buf out = {0};
    const char *p = (const char *)data;
    size_t len = size;
    /* run every complete command in the input, as the server would */
    for (;;) {
        kv_cmd cmd;
        size_t used = 0;
        if (kv_parse(p, len, &cmd, &used) != KV_PARSE_OK) {
            break;
        }
        kv_exec(&cmd, &store, &out);
        p += used;
        len -= used;
    }
    kv_buf_free(&out);
    kv_store_free(&store);
    return 0;
}

/* One request, with the arguments copied out. The bulk length is trusted: nothing
 * checks it against the room in the argument. A length of 40 writes 40 bytes into 32,
 * and the attacker chooses the 8 that land on whatever the heap keeps next. */
int trusting_parse(const char *data, size_t len, trusting_cmd *cmd) {
    const char *end = data + len;
    const char *p = data;
    const char *eol = find_eol(p, end);
    if (eol == NULL || *p != '*') {
        return -1;
    }
    long n;
    if (!parse_number(p + 1, eol, &n) || n > TRUSTING_MAX_ARGS) {
        return -1;
    }
    p = eol + 1;
    for (long i = 0; i < n; i++) {
        if (p >= end || *p != '$' || (eol = find_eol(p, end)) == NULL) {
            return -1;
        }
        long blen;
        if (!parse_number(p + 1, eol, &blen)) {
            return -1;
        }
        p = eol + 1;
        if (end - p < blen + 2) {
            return -1;
        }
        memcpy(cmd->argv[i], p, (size_t)blen); /* the missing check was here */
        cmd->arglen[i] = (size_t)blen;
        p += blen + 2;
    }
    cmd->argc = (size_t)n;
    return 0;
}

func TestReadCommand(t *testing.T) {
	cases := []struct {
		name  string
		input string
		want  []string
		err   error
	}{
		{"array", "*2\r\n$4\r\nECHO\r\n$2\r\nhi\r\n", []string{"ECHO", "hi"}, nil},
		{"inline", "  SET  a   b\r\n", []string{"SET", "a", "b"}, nil},
		{"bare LF", "PING\n", []string{"PING"}, nil},
		{"empty value", "*3\r\n$3\r\nSET\r\n$1\r\nk\r\n$0\r\n\r\n", []string{"SET", "k", ""}, nil},
		{"closed", "", nil, io.EOF},
		{"cut short", "*2\r\n$4\r\nEC", nil, io.ErrUnexpectedEOF},
		{"not a number", "*x\r\n", nil, ErrProtocol},
		{"wrong marker", "*1\r\n#4\r\nECHO\r\n", nil, ErrProtocol},
		{"too many", "*100\r\n", nil, ErrProtocol},
	}
	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {
			t.Parallel() // the cases share nothing, so they may run at once
			got, err := ReadCommand(bufio.NewReader(strings.NewReader(c.input)))
			if !errors.Is(err, c.err) {
				t.Fatalf("err = %v, want %v", err, c.err)
			}
			expectArgs(t, got, c.want)
		})
	}
}

// expectArgs compares the parsed arguments with the expected strings. t.Helper makes a
// failure report the caller's line, not this one.
func expectArgs(t *testing.T, got [][]byte, want []string) {
	t.Helper()
	if len(got) != len(want) {
		t.Fatalf("got %q, want %q", got, want)
	}
	for i := range got {
		if string(got[i]) != want[i] {
			t.Fatalf("arg %d = %q, want %q", i, got[i], want[i])
		}
	}
}

// The file is closed even when the sync failed, and the caller hears about both.
// errors.Join drops the nils, so with no error at all it returns nil.
return errors.Join(sync, l.f.Close())

#[cfg(test)]
mod tests {
    use super::*;
    use crate::store::MapStore;

    /// A command typed as one inline line, run against the store, and its reply as
    /// text. The helper is compiled only for tests, like the module around it.
    fn run(store: &mut MapStore, line: &str) -> String {
        let args: Command = line.split(' ').map(str::as_bytes).collect();
        let mut out = Vec::new();
        execute(&args, store, &mut out).unwrap();
        String::from_utf8(out).unwrap()
    }

    #[test]
    fn incr_counts_from_nothing() {
        let mut store = MapStore::new();
        assert_eq!(run(&mut store, "INCR n"), ":1\r\n");
        assert_eq!(run(&mut store, "INCR n"), ":2\r\n");
        assert_eq!(run(&mut store, "GET n"), "$1\r\n2\r\n");
    }

    #[test]
    fn an_expiry_in_the_past_deletes_the_key() {
        let mut store = MapStore::new();
        run(&mut store, "SET k v");
        assert_eq!(run(&mut store, "EXPIRE k 0"), ":1\r\n");
        assert_eq!(run(&mut store, "EXISTS k"), ":0\r\n");
    }
}

/// One request from the front of `buf`: an array of bulk strings such as
/// `*2\r\n$4\r\nECHO\r\n$2\r\nhi\r\n`, or one inline line of words. The returned
/// `Command` borrows `buf`, which the signature says with the elided lifetime: one
/// reference in, so the one reference out lives as long as it.
///
/// A code block in a doc comment is a test. `cargo test` compiles and runs it, and
/// `cargo doc` shows it, so an example that stops being true stops the build:
///
/// ```
/// use kv::resp::{Parsed, parse};
/// let Ok(Parsed::Done(args, used)) = parse(b"*2\r\n$4\r\nECHO\r\n$2\r\nhi\r\n") else {
///     unreachable!("a whole request")
/// };
/// assert_eq!(args, [&b"ECHO"[..], b"hi"]);
/// assert_eq!(used, 22);
/// assert!(matches!(parse(b"*2\r\n$4\r\nEC"), Ok(Parsed::Incomplete)));
/// ```
pub fn parse(buf: &[u8]) -> Result<Parsed<'_>> {
    let Some((line, mut used)) = take_line(buf, 0) else {
        return Ok(Parsed::Incomplete);
    };
    if line.first() != Some(&b'*') {
        let words = line.split(|b| *b == b' ').filter(|w| !w.is_empty());
        return Ok(Parsed::Done(words.collect(), used));
    }
    let n = parse_len(&line[1..], b'*')?;
    if n > MAX_ARGS {
        return Err(KvError::Protocol(b'*'));
    }
    let mut args = Vec::with_capacity(n);
    for _ in 0..n {
        let Some((line, start)) = take_line(buf, used) else {
            return Ok(Parsed::Incomplete);
        };
        if line.first() != Some(&b'$') {
            return Err(KvError::Protocol(line.first().copied().unwrap_or(b'\n')));
        }
        let size = parse_len(&line[1..], b'$')?;
        if size > MAX_BULK {
            return Err(KvError::Protocol(b'$'));
        }
        let end = start + size;
        if buf.len() < end + 2 {
            return Ok(Parsed::Incomplete);
        }
        if &buf[end..end + 2] != b"\r\n" {
            return Err(KvError::Protocol(buf[end]));
        }
        args.push(&buf[start..end]);
        used = end + 2;
    }
    Ok(Parsed::Done(args, used))
}

/// The line that starts at `from`, without its ending, and where the next line starts.
/// A bare LF is accepted as well as CRLF, so that a command typed into nc works.
fn take_line(buf: &[u8], from: usize) -> Option<(&[u8], usize)> {
    let rest = &buf[from..];
    let nl = rest.iter().position(|b| *b == b'\n')?;
    let line = rest[..nl].strip_suffix(b"\r").unwrap_or(&rest[..nl]);
    Some((line, from + nl + 1))
}

fn parse_len(digits: &[u8], type_byte: u8) -> Result<usize> {
    std::str::from_utf8(digits)
        .ok()
        .and_then(|s| s.parse().ok())
        .ok_or(KvError::Protocol(type_byte))
}

/// One request through the parser and the executor, as the server runs it, and the
/// reply bytes. A helper that every test in the file shares is an ordinary function.
fn round_trip(store: &mut MapStore, request: &[u8]) -> Vec<u8> {
    let mut out = Vec::new();
    match resp::parse(request) {
        Ok(Parsed::Done(args, used)) => {
            assert_eq!(used, request.len(), "the whole request is used");
            cmd::execute(&args, store, &mut out).expect("a Vec never fails to write");
        }
        other => panic!("not a whole request: {other:?}"),
    }
    out
}

#[test]
fn set_then_get() {
    let mut store = MapStore::new();
    assert_eq!(
        round_trip(&mut store, b"*3\r\n$3\r\nSET\r\n$1\r\nk\r\n$2\r\nhi\r\n"),
        b"+OK\r\n"
    );
    assert_eq!(
        round_trip(&mut store, b"*2\r\n$3\r\nGET\r\n$1\r\nk\r\n"),
        b"$2\r\nhi\r\n"
    );
}

#[test]
fn the_wrong_kind_is_an_error_and_not_a_panic() {
    let mut store = MapStore::new();
    round_trip(&mut store, b"*3\r\n$5\r\nLPUSH\r\n$1\r\nl\r\n$1\r\nx\r\n");
    let reply = round_trip(&mut store, b"*2\r\n$3\r\nGET\r\n$1\r\nl\r\n");
    assert!(
        reply.starts_with(b"-WRONGTYPE"),
        "{}",
        String::from_utf8_lossy(&reply)
    );
}

#[test]
fn an_unknown_command_is_named_in_the_error() {
    let mut store = MapStore::new();
    assert_eq!(
        round_trip(&mut store, b"*1\r\n$4\r\nNOPE\r\n"),
        b"-ERR unknown command 'NOPE'\r\n"
    );
}

fuzz_target!(|data: &[u8]| {
    let mut input = data;
    let mut store = kv::store::MapStore::new();
    let mut out = Vec::new();
    // run every complete command in the input, as the server would
    while let Ok(kv::resp::Parsed::Done(args, used)) = kv::resp::parse(input) {
        let _ = kv::cmd::execute(&args, &mut store, &mut out);
        input = &input[used..];
    }
});

/* Without restrict, the compiler must allow for dst and src overlapping, so each store
 * may change the next load, and the loop is kept in order or guarded by a check. With
 * the promise, the loop may add eight or sixteen numbers in one instruction. */
static void add_plain(int32_t *dst, const int32_t *src, size_t n) {
    for (size_t i = 0; i < n; i++) {
        dst[i] += src[i];
    }
}

static void add_restrict(int32_t *restrict dst, const int32_t *restrict src, size_t n) {
    for (size_t i = 0; i < n; i++) {
        dst[i] += src[i];
    }
}

// BenchmarkReadCommand times the parser on one SET request. go test -bench runs the loop
// for as long as it needs to get a stable number. The reader is reused, so that the
// number is the parser and not the cost of a 4 KB buffer per request.
func BenchmarkReadCommand(b *testing.B) {
	req := []byte("*3\r\n$3\r\nSET\r\n$16\r\nkey:000000001234\r\n$3\r\nxxx\r\n")
	r := bufio.NewReader(nil)
	for b.Loop() {
		r.Reset(bytes.NewReader(req))
		if _, err := ReadCommand(r); err != nil {
			b.Fatal(err)
		}
	}
}

// readLine returns one line without its line ending. ReadSlice hands back a view into
// the reader's own buffer, valid until the next read, so no allocation is made for a
// line. Chapter 13 measured the difference. A line longer than the buffer is an error,
// which for a length field it should be.
func readLine(r *bufio.Reader) ([]byte, error) {
	line, err := r.ReadSlice('\n')
	if err != nil {
		if err == io.EOF && len(line) > 0 {
			return nil, io.ErrUnexpectedEOF
		}
		if err == bufio.ErrBufferFull {
			return nil, protoErr(line[0])
		}
		return nil, err
	}
	line = bytes.TrimSuffix(line, []byte("\n"))
	return bytes.TrimSuffix(line, []byte("\r")), nil
}

// atoi parses a non-negative decimal from bytes without making a string of them, which
// strconv.Atoi would need and the compiler would allocate.
func atoi(b []byte) (int, bool) {
	if len(b) == 0 || len(b) > 18 {
		return 0, false
	}
	n := 0
	for _, c := range b {
		if c < '0' || c > '9' {
			return 0, false
		}
		n = n*10 + int(c-'0')
	}
	return n, true
}

/// Every allocation in the program goes through the global allocator, and a program may
/// name its own. This one counts the calls and passes them on to the system allocator.
struct Counting;

static ALLOCS: AtomicU64 = AtomicU64::new(0);

// SAFETY: every call is passed through to System unchanged, which is itself a correct
// allocator, and the counter has no effect on the memory handed back.
unsafe impl GlobalAlloc for Counting {
    unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
        ALLOCS.fetch_add(1, Ordering::Relaxed);
        unsafe { System.alloc(layout) }
    }

    unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
        unsafe { System.dealloc(ptr, layout) }
    }
}

#[global_allocator]
static GLOBAL: Counting = Counting;

fn main() {
    let req: &[u8] = b"*3\r\n$3\r\nSET\r\n$16\r\nkey:000000001234\r\n$3\r\nxxx\r\n";
    let runs = 10_000_000u32;
    let mut ok = 0u32;
    let before = ALLOCS.load(Ordering::Relaxed);
    let start = Instant::now();
    for _ in 0..runs {
        // the parser takes a slice and copies nothing, so no allocation is made per argument
        ok += u32::from(matches!(
            kv::resp::parse(req),
            Ok(kv::resp::Parsed::Done(..))
        ));
    }
    let elapsed = start.elapsed();
    let allocs = ALLOCS.load(Ordering::Relaxed) - before;
    println!(
        "{runs} parses in {} ms: {:.1} ns each ({ok} ok), {} allocations a parse",
        elapsed.as_millis(),
        elapsed.as_nanos() as f64 / f64::from(runs),
        allocs / u64::from(runs)
    );
}

# Keep function names in the release build, so a profiler can show them. The code is
# as fast as without this; only the binary is larger.
[profile.release]
debug = 1

/// A reply as a value, so that a command can be run under a lock and written after
/// the lock is released. Formatting and the write to the socket cost more than the
/// store operation itself, and neither needs the store.
pub enum Reply {
    Simple(&'static str),
    Error(String),
    Bulk(Vec<u8>),
    Int(i64),
    Nil,
    Array(Vec<Reply>),
}

pub fn write_reply(w: &mut (impl Write + ?Sized), reply: &Reply) -> io::Result<()> {
    match reply {
        Reply::Simple(s) => write_simple(w, s),
        Reply::Error(s) => write_error(w, s),
        Reply::Bulk(b) => write_bulk(w, b),
        Reply::Int(n) => write_int(w, *n),
        Reply::Nil => write_nil(w),
        Reply::Array(items) => {
            write_array_len(w, items.len())?;
            items.iter().try_for_each(|item| write_reply(w, item))
        }
    }
}

// A read takes the shared lock, which many threads hold at once. A
// write takes the exclusive one. Either way the lock is released
// before the reply is formatted or written.
let reply = {
    let db = db.read().unwrap_or_else(|e| e.into_inner());
    db.execute_read(&args)
};
let reply = match reply {
    Some(reply) => reply,
    None => {
        let mut db = db.write().unwrap_or_else(|e| e.into_inner());
        db.execute(&args)?
    }
};
resp::write_reply(&mut writer, &reply)?;

/* The settings, from the command line first and the environment second. */
typedef struct {
    int port;
    const char *aof;   /* NULL for no persistence */
    const char *fsync; /* NULL for the default */
    const char *hash;  /* "fnv" for the unseeded hash */
} kv_options;

static const char USAGE[] = "usage: kv [-p port] [-a file] [-f always|everysec|no]\n"
                            "  -p port   the port to listen on (KV_PORT), 6380 by default\n"
                            "  -a file   append every write to this file (KV_AOF)\n"
                            "  -f policy how often to fsync the file (KV_FSYNC)\n";

/* getopt walks argv for the options in its string. A letter followed by a colon takes an
 * argument, which getopt leaves in optarg. It returns -1 when the options are done, and
 * optind then says where the rest of the arguments start. */
static kv_options kv_options_parse(int argc, char **argv) {
    const char *port = getenv("KV_PORT");
    kv_options o = {
        .port = port && *port ? atoi(port) : 6380,
        .aof = getenv("KV_AOF"),
        .fsync = getenv("KV_FSYNC"),
        .hash = getenv("KV_HASH"),
    };
    int c;
    while ((c = getopt(argc, argv, "p:a:f:h")) != -1) {
        switch (c) {
        case 'p':
            o.port = atoi(optarg);
            break;
        case 'a':
            o.aof = optarg;
            break;
        case 'f':
            o.fsync = optarg;
            break;
        case 'h':
            fputs(USAGE, stdout);
            exit(0);
        default: /* getopt has already printed which option it did not know */
            die("%s", USAGE);
        }
    }
    if (optind < argc) {
        die("kv: unexpected argument '%s'", argv[optind]);
    }
    if (o.port <= 0 || o.port > 65535) {
        die("kv: not a port: %d", o.port);
    }
    if (o.aof != NULL && *o.aof == '\0') {
        o.aof = NULL;
    }
    return o;
}

/* Print a message the way printf would, and exit. _Noreturn tells the compiler, and
 * the reader, that nothing follows a call. The attribute is a GNU extension that gcc
 * and clang share, outside the standard: it checks the arguments against the format
 * string at compile time, as the compiler does for printf itself. */
__attribute__((format(printf, 1, 2))) _Noreturn static void die(const char *fmt, ...) {
    va_list ap;
    va_start(ap, fmt);
    vfprintf(stderr, fmt, ap);
    va_end(ap);
    fputc('\n', stderr);
    exit(2);
}

/* nullptr has a type of its own, so it cannot be mistaken for an int, as 0 or NULL can */
static const char *first_or(const char *s, const char *fallback) {
    return s != nullptr ? s : fallback;
}

/* [[nodiscard]] makes the compiler warn when a caller ignores the result */
[[nodiscard]] static int parse_digit(char c) {
    return c >= '0' && c <= '9' ? c - '0' : -1;
}

/* #embed pastes the bytes of a file into an array at compile time */
static const unsigned char banner[] = {
#embed "banner.txt"
};

int main(void) {
    /* constexpr: a true compile-time constant, usable where a #define was needed before */
    constexpr int limit = 64;
    /* typeof and auto: the type comes from the expression, as in the other two languages */
    typeof(limit) half = limit / 2;
    auto digit = parse_digit('7');
    /* binary literals and digit separators */
    unsigned mask = 0b1111'0000;
    /* true and false are keywords, and bool needs no header */
    bool ok = digit == 7;
    static_assert(limit > 0); /* no message needed any more */
    printf("%s %d %d %u %d %.*s", first_or(nullptr, "default"), half, digit, mask, ok,
           (int)sizeof banner, banner);
    return 0;
}

// config is every setting the server takes. The json tags name the keys in a config
// file, and the same struct is filled from the environment and the flags.
type config struct {
	Port  string `json:"port"`
	AOF   string `json:"aof"`
	Fsync string `json:"fsync"`
}

func env(name, fallback string) string {
	if v := os.Getenv(name); v != "" {
		return v
	}
	return fallback
}

// loadConfig reads the settings in order: the environment, then a JSON file if -config
// names one, then the flags given on the command line, each layer over the last.
func loadConfig() config {
	cfg := config{Port: env("KV_PORT", "6380"), AOF: os.Getenv("KV_AOF"), Fsync: os.Getenv("KV_FSYNC")}
	file := flag.String("config", "", "a JSON file with port, aof and fsync")
	flag.StringVar(&cfg.Port, "port", cfg.Port, "the port to listen on")
	flag.StringVar(&cfg.AOF, "aof", cfg.AOF, "append every write to this file")
	flag.StringVar(&cfg.Fsync, "fsync", cfg.Fsync, "how often to fsync the file: always, everysec or no")
	showVersion := flag.Bool("version", false, "print the version and exit")
	flag.Parse()
	if *showVersion {
		fmt.Print(version)
		os.Exit(0)
	}
	if *file != "" {
		data, err := os.ReadFile(*file)
		if err == nil {
			err = json.Unmarshal(data, &cfg)
		}
		if err != nil {
			slog.Error("config", "file", *file, "err", err)
			os.Exit(2)
		}
		// a flag given on the command line still wins over the file
		flag.Visit(func(f *flag.Flag) {
			switch f.Name {
			case "port":
				cfg.Port = f.Value.String()
			case "aof":
				cfg.AOF = f.Value.String()
			case "fsync":
				cfg.Fsync = f.Value.String()
			}
		})
	}
	return cfg
}

// go generate runs the command in the directive and writes version.txt, which is checked
// in. go:embed then compiles the file's contents into the binary as a string, so that a
// shipped kv can say which commit it is.
//
//go:generate sh -c "git describe --always --dirty > version.txt"
//go:embed version.txt
var version string

/// Every setting the server takes, from the environment first and the command line
/// second. The standard library gives an iterator over the arguments and nothing else.
/// Most Rust programs reach the same result with the clap crate, which writes the
/// parser and the help text from a struct like this one.
struct Options {
    port: u16,
    aof: Option<String>,
    fsync: String,
}

const USAGE: &str = "usage: kv [-p port] [-a file] [-f always|everysec|no]
  -p port   the port to listen on (KV_PORT), 6380 by default
  -a file   append every write to this file (KV_AOF)
  -f policy how often to fsync the file (KV_FSYNC)";

fn env(name: &str) -> Option<String> {
    std::env::var(name).ok().filter(|v| !v.is_empty())
}

/// A usage error is reported and the process ends with status 2, as getopt programs do.
/// The return type `!` is Rust's _Noreturn: this function never comes back.
fn usage_error(msg: &str) -> ! {
    eprintln!("kv: {msg}\n{USAGE}");
    std::process::exit(2)
}

fn options() -> Options {
    let mut o = Options {
        port: 6380,
        aof: env("KV_AOF"),
        fsync: env("KV_FSYNC").unwrap_or_default(),
    };
    let mut port = env("KV_PORT").unwrap_or_else(|| "6380".to_string());
    let mut args = std::env::args().skip(1);
    while let Some(flag) = args.next() {
        let mut value = || {
            args.next()
                .unwrap_or_else(|| usage_error(&format!("{flag} needs a value")))
        };
        match flag.as_str() {
            "-p" => port = value(),
            "-a" => o.aof = Some(value()).filter(|v| !v.is_empty()),
            "-f" => o.fsync = value(),
            "-h" => {
                println!("{USAGE}");
                std::process::exit(0)
            }
            other => usage_error(&format!("unknown option '{other}'")),
        }
    }
    o.port = match port.parse() {
        Ok(port) => port,
        Err(_) => usage_error(&format!("not a port: {port}")),
    };
    o
}