path: root/src/api/store.rs

//! /store and /stream handlers.

use std::io::Error as IoError;

use axum::{
    Extension,
    body::{Body, Bytes},
    extract::{Path, Query, RawQuery, State},
    http::{
        HeaderMap, HeaderValue, StatusCode,
        header::{ACCEPT, CONTENT_TYPE, ETAG, LOCATION, VARY},
    },
    response::{IntoResponse, Response},
};
use serde::Deserialize;
use serde_json::Value as JsonValue;
use sha2::{Digest, Sha256};
use tokio::sync::broadcast;
use tracing::{debug, error, instrument, warn};

use super::auth::AuthenticatedUser;
use super::{ApiError, store_status};
use crate::document::ChangeHash;
use crate::document::meta;
use crate::document::store::{self, StoreError, paths_related};
use crate::document::typed::{self, JsonForm, Kind, TimestampFormat, Val};
use crate::state::{AppState, ReadSource};
use crate::transport::Hex;

#[derive(Debug, Default, Deserialize)]
pub(crate) struct ReadQuery {
    #[serde(default)]
    conflicts: bool,
    /// Return the document's heads at the time of the read (`x-continuum-heads`), e.g.
    /// to base a replacing remake on exactly what was read.
    #[serde(default)]
    heads: bool,
    /// What's there, one level deep (`store::read_shallow`), to browse a large store a
    /// level at a time.
    #[serde(default)]
    shallow: bool,
    /// Read as the document was at these heads (comma-separated hex, as
    /// `x-continuum-heads` gives them), not as it is now (`document::at`).
    at: Option<String>,
}

impl ReadQuery {
    /// The heads `at` names, if any; `400` if it doesn't name them.
    fn at(&self) -> Result<Option<Vec<ChangeHash>>, ApiError> {
        let Some(at) = &self.at else {
            return Ok(None);
        };
        at.split(',')
            .map(|head| head.trim().parse::<ChangeHash>())
            .collect::<Result<Vec<_>, _>>()
            .map(Some)
            .map_err(|_| ApiError(StatusCode::BAD_REQUEST, Some("at: not heads".to_owned())))
    }
}

/// Reads `doc` with `read`, as it is or, with `at`, as it was at those heads
/// (`UnknownHeads` if it lacks one).
macro_rules! read_at {
    ($doc:expr, $at:expr, |$d:ident| $read:expr) => {
        match $at {
            None => {
                let $d = $doc;
                $read
            }
            Some(heads) => match crate::document::at::At::new($doc, heads) {
                Some(at) => {
                    let $d = &at;
                    $read
                }
                None => Err(From::from(StoreError::UnknownHeads)),
            },
        }
    };
}

#[derive(Debug, Deserialize)]
pub(crate) struct PostQuery {
    /// Adds this to the counter at the path instead of inserting the body
    /// (`store::increment_path`).
    increment: Option<i64>,
    /// Casts what is at the path to this type instead of inserting the body
    /// (`store::cast_path`; `typed::Kind::name` names it).
    cast: Option<String>,
    /// With `cast`, how timestamps are read from and written as strings (`rfc3339` by
    /// default; `typed::TimestampFormat`).
    format: Option<String>,
}

#[derive(Debug, Deserialize)]
pub(crate) struct DeleteQuery {
    #[serde(default)]
    prune: bool,
}

/// `GET /store`: the whole store (`docs/openapi.yaml`).
pub(crate) async fn get_store_root(
    State(state): State<AppState>,
    headers: HeaderMap,
    Query(query): Query<ReadQuery>,
) -> Result<Response, ApiError> {
    read(&state, Vec::new(), &headers, &query).await
}

/// `GET /store/{key}`.
pub(crate) async fn get_store_entry(
    State(state): State<AppState>,
    Path(key): Path<String>,
    headers: HeaderMap,
    Query(query): Query<ReadQuery>,
    RawQuery(raw_query): RawQuery,
) -> Result<Response, ApiError> {
    let path = parse_key_path(&key)?;
    let at = query.at()?;
    if let Some(redirect) =
        to_positive(&state, &path, false, at.as_deref(), raw_query.as_deref()).await?
    {
        return Ok(redirect);
    }
    read(&state, path, &headers, &query).await
}

/// A read of `path` as `query` asks (the empty path: the whole store). The root is read
/// while not ready too (`render_read`'s `unready_live`), and is never missing: an empty
/// store is an empty map.
async fn read(
    state: &AppState,
    path: Vec<String>,
    headers: &HeaderMap,
    query: &ReadQuery,
) -> Result<Response, ApiError> {
    let root = path.is_empty();
    if query.shallow {
        return shallow(state, path, root, query).await;
    }
    let accept = headers.get(ACCEPT).cloned();
    let want_conflicts = query.conflicts;
    let at = query.at()?;

    let (response, source, heads) = state
        .render_read(
            root,
            query.heads,
            move |doc| {
                read_at!(doc, at.as_deref(), |doc| read_shown(
                    doc,
                    &path,
                    accept.as_ref(),
                    want_conflicts
                ))
            },
            move |result| match result {
                Ok(Some(read)) => read.respond(),
                Ok(None) if root => Read {
                    held: store::Held {
                        kind: Kind::Map,
                        value: None,
                    },
                    shown: Shown::Json(JsonForm::default()),
                    json: Some(JsonValue::Object(Default::default())),
                    conflicts: want_conflicts.then(Vec::new),
                }
                .respond(),
                Ok(None) => Err(StatusCode::NOT_FOUND.into()),
                Err(error) => Err(error),
            },
        )
        .await?;
    mark_source(response, source, heads)
}

/// A value read, to show as asked (`Shown`).
struct Read {
    held: store::Held,
    shown: Shown,
    /// Its JSON, of the form asked, unless shown natively.
    json: Option<JsonValue>,
    /// With `?conflicts=true`: those on the way (then always JSON).
    conflicts: Option<Vec<store::ConflictReport>>,
}

impl Read {
    fn respond(self) -> Result<Response, ApiError> {
        if let Some(conflicts) = self.conflicts {
            let value = self.json.unwrap_or(JsonValue::Null);
            let body = serde_json::json!({ "value": value, "conflicts": conflicts });
            return show(&self.held, Shown::Json(JsonForm::default()), Some(&body));
        }
        show(&self.held, self.shown, self.json.as_ref())
    }
}

/// What `path` holds, read to show as `accept` asks; `406` if it can't be.
fn read_shown(
    doc: &impl crate::document::ReadDoc,
    path: &[String],
    accept: Option<&HeaderValue>,
    conflicts: bool,
) -> Result<Option<Read>, ApiError> {
    let failed = |e| store_status(e, "Failed to read store path");
    let Some(held) = store::read_held(doc, path).map_err(failed)? else {
        return Ok(None);
    };
    let shown = if conflicts {
        Shown::Json(json_form(accept)?)
    } else {
        shown(accept, held.kind)?
    };
    let (json, conflicts) = match (shown, conflicts) {
        (Shown::Native(_), _) => (None, None),
        (Shown::Json(form), true) => {
            match store::read_path_with_conflicts_as(doc, path, form).map_err(failed)? {
                Some((json, conflicts)) => (Some(json), Some(conflicts)),
                None => return Ok(None),
            }
        }
        (Shown::Json(form), false) => (store::read_path_as(doc, path, form).map_err(failed)?, None),
        (Shown::Text, _) => (
            store::read_path_as(doc, path, JsonForm::default()).map_err(failed)?,
            None,
        ),
    };
    Ok(Some(Read {
        held,
        shown,
        json,
        conflicts,
    }))
}

/// `?shallow=true`: what's at `path`, one level deep, as JSON (`store::read_shallow`);
/// with conflicts asked for too, `400`. `unready_live` as `render_read` has it.
async fn shallow(
    state: &AppState,
    path: Vec<String>,
    unready_live: bool,
    query: &ReadQuery,
) -> Result<Response, ApiError> {
    if query.conflicts {
        return Err(StatusCode::BAD_REQUEST.into());
    }
    let at = query.at()?;
    let (response, source, heads) = state
        .render_read(
            unready_live,
            query.heads,
            move |doc| read_at!(doc, at.as_deref(), |doc| store::read_shallow(doc, &path)),
            move |result| {
                let value = result
                    .map_err(|e| store_status(e, "Failed to read store path"))?
                    .ok_or(StatusCode::NOT_FOUND)?;
                respond(None, "application/json", json_bytes(&value)?)
            },
        )
        .await?;
    mark_source(response, source, heads)
}

/// Header on reads served from the pre-rejoin document; its value is that epoch.
const STALE_HEADER: &str = "x-continuum-stale";

/// Header with the document's heads at the time of the read (sorted, comma-separated
/// hex), on request (`?heads=true`); and on every write, after it: a read with these
/// (`?at=`) includes the write.
const HEADS_HEADER: &str = "x-continuum-heads";

fn with_heads(mut response: Response, heads: &[ChangeHash]) -> Response {
    let joined = heads
        .iter()
        .map(ToString::to_string)
        .collect::<Vec<_>>()
        .join(",");
    if let Ok(value) = HeaderValue::from_str(&joined) {
        response.headers_mut().insert(HEADS_HEADER, value);
    }
    response
}

fn mark_source(
    response: Result<Response, ApiError>,
    source: ReadSource,
    heads: Option<Vec<ChangeHash>>,
) -> Result<Response, ApiError> {
    let mut response = response?;
    if let Some(heads) = heads {
        response = with_heads(response, &heads);
    }
    if let ReadSource::Stale(epoch) = source
        && let Ok(value) = HeaderValue::from_str(&epoch.to_string())
    {
        response.headers_mut().insert(STALE_HEADER, value);
    }
    Ok(response)
}

#[derive(Debug, Deserialize)]
pub(crate) struct PutQuery {
    /// Whether the write may change the type of what it writes over (`true`, the
    /// default); if not, one that would is `409` (`store::Writing::retype`).
    #[serde(default = "yes")]
    retype: bool,
}

fn yes() -> bool {
    true
}

/// Header listing the places whose type a write changed (`store::Retyped`), as
/// `<path>: <from> -> <to>`, comma-separated, `MAX_RETYPED_LISTED` at most (then `…`).
const RETYPED_HEADER: &str = "x-continuum-retyped";
const MAX_RETYPED_LISTED: usize = 16;

/// `PUT /store/{key}`: answered with what is there after the write, shown as `Accept`
/// asks (read back: with sticky types, a JSON value may be stored as another type).
pub(crate) async fn put_store_entry(
    State(state): State<AppState>,
    Extension(user): Extension<AuthenticatedUser>,
    Path(key): Path<String>,
    Query(query): Query<PutQuery>,
    RawQuery(raw_query): RawQuery,
    headers: HeaderMap,
    body: Bytes,
) -> Result<Response, ApiError> {
    if !state.is_ready().await {
        return Err(StatusCode::SERVICE_UNAVAILABLE.into());
    }

    let path = parse_write_path(&state, &user, &key, "PUT")?;
    if let Some(redirect) = to_positive(&state, &path, false, None, raw_query.as_deref()).await? {
        return Ok(redirect);
    }
    let body = body_value(&headers, &body)?;
    let accept = headers.get(ACCEPT).cloned();
    let writing = store::Writing::new(body.sticky, body.timestamps, query.retype);

    // A value that can't be shown as asked is refused, and the write with it.
    let ((retyped, read), heads) = state
        .mutate_doc_for(Some(user.0.clone()), move |doc| {
            let retyped = store::write_val_path(doc, &path, &body.value, writing)
                .map_err(|e| store_status(e, "Failed to write store value"))?;
            let read = read_shown(&*doc, &path, accept.as_ref(), false)?;
            Ok::<_, ApiError>((retyped, read))
        })
        .await?;
    let read = read.ok_or(ApiError(StatusCode::INTERNAL_SERVER_ERROR, None))?;

    Ok(with_retyped(with_heads(read.respond()?, &heads), &retyped))
}

/// `response` saying where a write changed types (`RETYPED_HEADER`), if it did.
fn with_retyped(mut response: Response, retyped: &[store::Retyped]) -> Response {
    if !retyped.is_empty() {
        let mut listed: Vec<String> = retyped
            .iter()
            .take(MAX_RETYPED_LISTED)
            .map(ToString::to_string)
            .collect();
        if retyped.len() > MAX_RETYPED_LISTED {
            listed.push("…".to_owned());
        }
        if let Ok(value) = HeaderValue::from_str(&listed.join(", ")) {
            response.headers_mut().insert(RETYPED_HEADER, value);
        }
    }
    response
}

fn etag_for(bytes: &[u8]) -> String {
    format!("\"{}\"", Hex(&Sha256::digest(bytes)))
}

fn parse_key_path(key: &str) -> Result<Vec<String>, StatusCode> {
    let segments: Vec<String> = key.split('/').map(str::to_owned).collect();

    if segments.is_empty() || segments.iter().any(|segment| segment.is_empty()) {
        return Err(StatusCode::BAD_REQUEST);
    }

    Ok(segments)
}

/// A path for a write. Protocol data (`_continuum`: remake records, members, forget
/// records, epoch metadata) is what every node acts on, so a write there acts on the
/// whole swarm: only users listed in `--protocol-writers` may, and each such write is
/// logged. Epoch metadata (and `_continuum` as a whole, which contains it) is never
/// writable: the protocol reads it at the snapshot change, so an edit only misleads
/// whoever reads it later. `403` otherwise.
fn parse_write_path(
    state: &AppState,
    user: &AuthenticatedUser,
    key: &str,
    method: &str,
) -> Result<Vec<String>, StatusCode> {
    let path = parse_key_path(key)?;
    if !meta::is_protocol(&path) {
        return Ok(path);
    }
    // `_continuum` itself, epoch metadata and the swarm ID: nobody.
    let fixed = path.len() < 2 || path[1] == meta::EPOCHS || path[1] == meta::SWARM;
    if fixed || !state.protocol_writers.contains(&user.0) {
        warn!(user = %user.0, method, path = %key, "Refused a write to protocol data");
        return Err(StatusCode::FORBIDDEN);
    }
    warn!(user = %user.0, method, path = %key, "Protocol data written through the store API; this acts on the whole swarm");
    Ok(path)
}

/// How a value is shown, as the request's `Accept` asks.
#[derive(Clone, Copy, Debug)]
enum Shown {
    Json(JsonForm),
    /// `text/plain`: a string as it is, anything else as JSON.
    Text,
    /// Natively, as its type's own media type (`typed::native`).
    Native(TimestampFormat),
}

/// The JSON form `accept` asks for (`application/json; timestamp-format=…`): `400` for
/// a timestamp format there is none of.
fn json_form(accept: Option<&HeaderValue>) -> Result<JsonForm, ApiError> {
    for (media, params) in accepted(accept) {
        match media.as_str() {
            "application/json" => return json_form_of(params),
            typed::TYPED_JSON => {
                return Ok(JsonForm {
                    typed: true,
                    ..json_form_of(params)?
                });
            }
            _ => {}
        }
    }
    Ok(JsonForm::default())
}

fn json_form_of(params: &str) -> Result<JsonForm, ApiError> {
    let timestamps = match typed::param(params, "timestamp-format") {
        None => TimestampFormat::default(),
        Some(name) => TimestampFormat::named(&name).ok_or_else(|| {
            ApiError(
                StatusCode::BAD_REQUEST,
                Some(format!("no timestamp format {name:?}")),
            )
        })?,
    };
    Ok(JsonForm {
        timestamps,
        typed: false,
    })
}

/// The media types of `accept`, in its order, each with its parameters (q-values are not
/// weighed). Without an `Accept`, none.
fn accepted(accept: Option<&HeaderValue>) -> Vec<(String, &str)> {
    let Some(accept) = accept.and_then(|value| value.to_str().ok()) else {
        return Vec::new();
    };
    accept
        .split(',')
        .map(|item| {
            let (media, params) = item.split_once(';').unwrap_or((item, ""));
            (media.trim().to_ascii_lowercase(), params)
        })
        .collect()
}

/// How a value of `kind` is shown, as `accept` asks: the first it names that the value
/// can be shown as. JSON without an `Accept`, or for `*/*`; `406` naming what it can be
/// shown as if `accept` names none of them.
fn shown(accept: Option<&HeaderValue>, kind: Kind) -> Result<Shown, ApiError> {
    let items = accepted(accept);
    if items.is_empty() {
        return Ok(Shown::Json(JsonForm::default()));
    }
    let native = kind.media_type();
    for (media, params) in &items {
        match media.as_str() {
            "application/json" => return Ok(Shown::Json(json_form_of(params)?)),
            typed::TYPED_JSON => {
                return Ok(Shown::Json(JsonForm {
                    typed: true,
                    ..json_form_of(params)?
                }));
            }
            typed::TEXT => return Ok(Shown::Text),
            "*/*" | "application/*" => return Ok(Shown::Json(JsonForm::default())),
            media if Some(media) == native => {
                let format = match typed::param(params, "format") {
                    None => TimestampFormat::default(),
                    Some(name) => TimestampFormat::named(&name).ok_or_else(|| {
                        ApiError(
                            StatusCode::BAD_REQUEST,
                            Some(format!("no timestamp format {name:?}")),
                        )
                    })?,
                };
                return Ok(Shown::Native(format));
            }
            _ => {}
        }
    }
    let mut offered = vec!["application/json", typed::TYPED_JSON, typed::TEXT];
    if let Some(native) = native.filter(|native| *native != typed::TEXT) {
        offered.push(native);
    }
    Err(ApiError(
        StatusCode::NOT_ACCEPTABLE,
        Some(format!(
            "a {} is shown as {}",
            kind.name(),
            offered.join(", ")
        )),
    ))
}

/// The response showing `held` as `shown`: `json` is its JSON, of the form `shown` asks,
/// where it isn't shown natively.
fn show(held: &store::Held, shown: Shown, json: Option<&JsonValue>) -> Result<Response, ApiError> {
    let json = || json.ok_or(ApiError(StatusCode::INTERNAL_SERVER_ERROR, None));
    match shown {
        Shown::Json(form) => {
            let media = if form.typed {
                typed::TYPED_JSON
            } else {
                "application/json"
            };
            respond(Some(held.kind), media, json_bytes(json()?)?)
        }
        Shown::Text => {
            let body = match json()? {
                JsonValue::String(value) => value.clone().into_bytes(),
                other => json_bytes(other)?,
            };
            respond(Some(held.kind), "text/plain; charset=utf-8", body)
        }
        Shown::Native(format) => {
            let (content_type, body) = held
                .value
                .as_ref()
                .and_then(|value| typed::native(value, format))
                .ok_or(ApiError(StatusCode::INTERNAL_SERVER_ERROR, None))?;
            respond(Some(held.kind), &content_type, body)
        }
    }
}

fn json_bytes(value: &JsonValue) -> Result<Vec<u8>, ApiError> {
    serde_json::to_vec(value).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR.into())
}

/// A response of `body` as `content_type`, with its `ETag`, and what the value is
/// stored as (`x-continuum-type`), if it is one.
fn respond(kind: Option<Kind>, content_type: &str, body: Vec<u8>) -> Result<Response, ApiError> {
    let etag = etag_for(&body);
    let mut response = body.into_response();
    let headers = response.headers_mut();
    headers.insert(
        CONTENT_TYPE,
        HeaderValue::from_str(content_type).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?,
    );
    headers.insert(
        ETAG,
        HeaderValue::from_str(&etag).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?,
    );
    if let Some(kind) = kind {
        headers.insert(TYPE_HEADER, HeaderValue::from_static(kind.name()));
        headers.insert(VARY, HeaderValue::from_static("Accept"));
    }
    Ok(response)
}

/// Header naming what a value read or written is stored as (`typed::Kind::name`).
const TYPE_HEADER: &str = "x-continuum-type";

/// A write's body: the value to write (`typed::Val`), from JSON (the default), typed
/// JSON, text as a string, or a scalar of a type of its own (`typed::from_native`).
struct WriteBody {
    value: Val,
    /// Which of its values keep the types of what they are written over
    /// (`store::Sticky`): plain JSON's and text's where they can, typed JSON's numbers
    /// over floats.
    sticky: store::Sticky,
    /// How timestamps are written in it (`timestamp-format`).
    timestamps: TimestampFormat,
}

fn body_value(headers: &HeaderMap, body: &Bytes) -> Result<WriteBody, ApiError> {
    let content_type = headers
        .get(CONTENT_TYPE)
        .and_then(|value| value.to_str().ok())
        .unwrap_or("application/json");
    let (media, params) = content_type.split_once(';').unwrap_or((content_type, ""));
    let media = media.trim().to_ascii_lowercase();
    let bad = |error: String| ApiError(StatusCode::BAD_REQUEST, Some(error));
    let json = || {
        serde_json::from_slice::<JsonValue>(body).map_err(|error| bad(format!("not JSON: {error}")))
    };
    let timestamps = json_form_of(params)?.timestamps;
    let (value, sticky) = match media.as_str() {
        "application/json" => (Val::from_json(&json()?).map_err(bad)?, store::Sticky::All),
        typed::TYPED_JSON => (
            Val::from_typed_json(&json()?, timestamps).map_err(bad)?,
            store::Sticky::Floats,
        ),
        typed::TEXT => (
            String::from_utf8(body.to_vec())
                .map(|text| Val::Scalar(automerge::ScalarValue::Str(text.into())))
                .map_err(|_| bad("not UTF-8".to_owned()))?,
            store::Sticky::All,
        ),
        media => match typed::from_native(media, params, body) {
            Some(Ok(scalar)) => (Val::Scalar(scalar), store::Sticky::No),
            Some(Err(error)) => return Err(bad(error)),
            None => return Err(StatusCode::UNSUPPORTED_MEDIA_TYPE.into()),
        },
    };
    Ok(WriteBody {
        value,
        sticky,
        timestamps,
    })
}

/// `POST /store/{key}`: inserts the body into a list, `key` ending in the index it
/// goes to (`store::insert_path`: the items from there on move up; the length, or `-1`,
/// appends); `201` with where it went (`{"index": n}`). Not a list there, or an index
/// past its end: `409`, saying which. With `?increment=n`, adds `n` to the counter at
/// `key` instead (`store::increment_path`), answering with its value then; not a
/// counter there: `409`. With `?cast=<type>`, casts what is at `key` to that type
/// instead (`store::cast_path`), answering with it then, as `PUT` does; no cast
/// possible: `409`.
pub(crate) async fn post_store_entry(
    State(state): State<AppState>,
    Extension(user): Extension<AuthenticatedUser>,
    Path(key): Path<String>,
    Query(query): Query<PostQuery>,
    RawQuery(raw_query): RawQuery,
    headers: HeaderMap,
    body: Bytes,
) -> Result<Response, ApiError> {
    if !state.is_ready().await {
        return Err(StatusCode::SERVICE_UNAVAILABLE.into());
    }
    let path = parse_write_path(&state, &user, &key, "POST")?;
    if query.increment.is_some() && query.cast.is_some() {
        return Err(ApiError(
            StatusCode::BAD_REQUEST,
            Some("increment or cast, not both".to_owned()),
        ));
    }
    // An increment or a cast names an item (`-1`: the last); an insert where it goes
    // (`-1`: the end).
    let insertion = query.increment.is_none() && query.cast.is_none();
    if let Some(redirect) =
        to_positive(&state, &path, insertion, None, raw_query.as_deref()).await?
    {
        return Ok(redirect);
    }
    if let Some(by) = query.increment {
        let (value, heads) = state
            .mutate_doc_for(Some(user.0.clone()), move |doc| {
                store::increment_path(doc, &path, by)
            })
            .await
            .map_err(|e| store_status(e, "Failed to increment a counter"))?;
        return Ok(with_heads(
            {
                let counter = automerge::ScalarValue::counter(value);
                let held = store::Held {
                    kind: Kind::Counter,
                    value: Some(counter),
                };
                show(
                    &held,
                    shown(headers.get(ACCEPT), Kind::Counter)?,
                    Some(&JsonValue::from(value)),
                )?
            },
            &heads,
        ));
    }
    if let Some(name) = &query.cast {
        let bad = |message: String| ApiError(StatusCode::BAD_REQUEST, Some(message));
        let to = Kind::named(name).ok_or_else(|| bad(format!("no type {name:?}")))?;
        let timestamps = match &query.format {
            None => TimestampFormat::default(),
            Some(format) => TimestampFormat::named(format)
                .ok_or_else(|| bad(format!("no timestamp format {format:?}")))?,
        };
        let accept = headers.get(ACCEPT).cloned();
        // A value that can't be shown as asked is refused, and the cast with it.
        let ((retyped, read), heads) = state
            .mutate_doc_for(Some(user.0.clone()), move |doc| {
                let retyped = store::cast_path(doc, &path, to, timestamps)
                    .map_err(|e| store_status(e, "Failed to cast a store value"))?;
                let read = read_shown(&*doc, &path, accept.as_ref(), false)?;
                Ok::<_, ApiError>((retyped, read))
            })
            .await?;
        let read = read.ok_or(ApiError(StatusCode::INTERNAL_SERVER_ERROR, None))?;
        return Ok(with_retyped(with_heads(read.respond()?, &heads), &retyped));
    }
    let index = path.last().and_then(|leaf| store::list_index(leaf));
    let value = body_value(&headers, &body)?;
    let inserted = state
        .mutate_doc_for(Some(user.0.clone()), move |doc| {
            store::insert_val_path(doc, &path, &value.value)
        })
        .await;
    match inserted {
        Ok(((), heads)) => Ok(with_heads(
            (
                StatusCode::CREATED,
                axum::Json(serde_json::json!({ "index": index })),
            )
                .into_response(),
            &heads,
        )),
        Err(error) => Err(store_status(error, "Failed to insert into a list")),
    }
}

/// `DELETE /store/{key}`.
#[instrument(skip(state), fields(path, prune = query.prune))]
pub(crate) async fn delete_store_entry(
    State(state): State<AppState>,
    Extension(user): Extension<AuthenticatedUser>,
    Path(key): Path<String>,
    Query(query): Query<DeleteQuery>,
    RawQuery(raw_query): RawQuery,
) -> Result<Response, ApiError> {
    let path = parse_write_path(&state, &user, &key, "DELETE")?;

    tracing::Span::current().record("path", path.join("/").as_str());

    if !state.is_ready().await {
        return Err(StatusCode::SERVICE_UNAVAILABLE.into());
    }

    if path.is_empty() {
        return Err(StatusCode::BAD_REQUEST.into());
    }

    if let Some(redirect) = to_positive(&state, &path, false, None, raw_query.as_deref()).await? {
        return Ok(redirect);
    }

    let prune = query.prune;
    let ((), heads) = state
        .mutate_doc_for(Some(user.0.clone()), move |doc| {
            store::delete_path(doc, &path, prune)
        })
        .await
        .map_err(|e| store_status(e, "Failed to delete store value"))?;

    Ok(with_heads(StatusCode::NO_CONTENT.into_response(), &heads))
}

/// A path with negative list indexes (`-1` the last item; for an insert, the end): a
/// `307` to it with them made positive (`store::positive_path`), the query kept, so a
/// client reads, writes or deletes what the index names now. None without any.
async fn to_positive(
    state: &AppState,
    path: &[String],
    insertion: bool,
    at: Option<&[ChangeHash]>,
    query: Option<&str>,
) -> Result<Option<Response>, ApiError> {
    if !path.iter().any(|segment| segment.starts_with('-')) {
        return Ok(None);
    }
    let positive = {
        let replica = state.doc.read().await;
        // Counted from the end of the list as it was then, for a read at heads.
        read_at!(&replica.doc, at, |doc| store::positive_path(
            doc, path, insertion
        ))
    }
    .map_err(|e| store_status(e, "Failed to resolve a negative index"))?;
    let Some(positive) = positive else {
        return Ok(None);
    };
    let mut location = format!(
        "/store/{}",
        positive
            .iter()
            .map(|segment| encode_segment(segment))
            .collect::<Vec<_>>()
            .join("/")
    );
    if let Some(query) = query.filter(|query| !query.is_empty()) {
        location.push('?');
        location.push_str(query);
    }
    let location = HeaderValue::from_str(&location).map_err(|_| StatusCode::BAD_REQUEST)?;
    Ok(Some(
        (StatusCode::TEMPORARY_REDIRECT, [(LOCATION, location)]).into_response(),
    ))
}

/// `segment` as a URL path segment: unreserved characters as they are, the rest
/// percent-encoded (UTF-8).
fn encode_segment(segment: &str) -> String {
    let mut encoded = String::with_capacity(segment.len());
    for byte in segment.bytes() {
        if byte.is_ascii_alphanumeric() || b"-._~".contains(&byte) {
            encoded.push(char::from(byte));
        } else {
            encoded.push_str(&format!("%{byte:02X}"));
        }
    }
    encoded
}

// --- streams -----------------------------------------------------------------

async fn render_traced(
    state: &AppState,
    path: Vec<String>,
) -> Result<(Option<JsonValue>, bool), StoreError> {
    state
        .render(
            move |doc| store::read_path_traced(doc, &path),
            |result| result,
        )
        .await
}

/// `GET /stream`: follows the whole store, as `GET /store` reads it (events have an
/// empty `key`).
pub(crate) async fn stream_store_root(State(state): State<AppState>) -> Result<Response, ApiError> {
    stream(state, String::new(), Vec::new()).await
}

/// `GET /stream/{key}`: follows one value.
pub(crate) async fn stream_store_entry(
    State(state): State<AppState>,
    Path(key): Path<String>,
) -> Result<Response, ApiError> {
    let path = parse_key_path(&key)?;
    stream(state, key, path).await
}

/// Newline-delimited JSON events for the value at `path` (`key` as the client wrote it):
/// the current value, then one whenever the resolved value changes.
async fn stream(state: AppState, key: String, path: Vec<String>) -> Result<Response, ApiError> {
    if !state.is_ready().await {
        return Err(StatusCode::SERVICE_UNAVAILABLE.into());
    }

    // Subscribe first, so nothing between the initial read and the loop is missed.
    let mut changes = state.change_events.subscribe();

    let initial = render_traced(&state, path.clone())
        .await
        .map_err(|e| store_status(e, "Failed to read initial stream value"))?;

    let stream_state = state.clone();
    let stream_key = key.clone();

    let output = async_stream::stream! {
        let (mut last_value, mut conflicted) = initial;

        match stream_event_line(&stream_key, last_value.as_ref()) {
            Ok(line) => yield Ok::<Bytes, IoError>(line),
            Err(error) => {
                yield Err(error);
                return;
            }
        }

        loop {
            let relevant = match changes.recv().await {
                // Conflicted paths can change without a patch on their path; re-read always.
                Ok(set) => conflicted || set.paths.iter().any(|changed| paths_related(changed, &path)),
                Err(broadcast::error::RecvError::Lagged(skipped)) => {
                    debug!(skipped, key = %stream_key, "Stream lagged; re-reading");
                    true
                }
                Err(broadcast::error::RecvError::Closed) => break,
            };

            if !relevant {
                continue;
            }

            let (next_value, next_conflicted) = match render_traced(&stream_state, path.clone()).await {
                Ok(result) => result,
                Err(error) => {
                    error!(error = %format!("{error:#}"), key = %stream_key, "Failed to read updated stream value");
                    yield Err(IoError::other("Failed to read updated store value"));
                    return;
                }
            };
            conflicted = next_conflicted;

            // A related change can still leave the resolved value unchanged.
            if next_value == last_value {
                continue;
            }

            match stream_event_line(&stream_key, next_value.as_ref()) {
                Ok(line) => yield Ok(line),
                Err(error) => {
                    yield Err(error);
                    return;
                }
            }
            last_value = next_value;
        }
    };

    let mut response = Body::from_stream(output).into_response();
    response.headers_mut().insert(
        CONTENT_TYPE,
        HeaderValue::from_static("application/x-ndjson"),
    );
    response.headers_mut().insert(
        axum::http::header::CACHE_CONTROL,
        HeaderValue::from_static("no-cache, no-transform"),
    );
    // Through nginx too as it comes, not buffered (`operations.md`).
    response.headers_mut().insert(
        super::UNBUFFERED.0,
        HeaderValue::from_static(super::UNBUFFERED.1),
    );
    Ok(response)
}

fn stream_event_line(key: &str, value: Option<&JsonValue>) -> std::io::Result<Bytes> {
    let mut event = serde_json::Map::new();

    event.insert("key".to_owned(), JsonValue::String(key.to_owned()));

    match value {
        Some(value) => {
            event.insert("value".to_owned(), value.clone());
        }
        None => {
            event.insert("deleted".to_owned(), JsonValue::Bool(true));
        }
    }

    let mut line = serde_json::to_vec(&JsonValue::Object(event)).map_err(IoError::other)?;

    line.push(b'\n');
    Ok(Bytes::from(line))
}