//! /store and /stream handlers.

use crate::document::signatures::Sign;
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::access::{Access, Before, Visible, prune, prune_shallow, registry_kept};
use super::auth::AuthenticatedUser;
use super::{ApiError, store_status};
use crate::access::envelope;
use crate::document::meta;
use crate::document::store::{self, StoreError, paths_related};
use crate::document::typed::{self, JsonForm, Kind, TimestampFormat, Val};
use crate::document::{ChangeHash, ScalarValue};
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>,
    /// Values their users signed as they are stored, envelopes (`access::envelope`), not
    /// opened (`opened`).
    #[serde(default)]
    raw: bool,
}

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>,
    /// Whether the change is signed whatever its path (`document::signatures`).
    #[serde(default)]
    signed: bool,
}

#[derive(Debug, Deserialize)]
pub(crate) struct DeleteQuery {
    #[serde(default)]
    prune: bool,
    /// Whether the change is signed whatever its path (`document::signatures`).
    #[serde(default)]
    signed: bool,
}

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

/// `GET /store/{key}`.
pub(crate) async fn get_store_entry(
    State(state): State<AppState>,
    Extension(access): Extension<Access>,
    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, access).await
}

/// `403` where the user may read nothing of `path` (`Access::visible`).
fn readable(access: &Access, path: &[String]) -> Result<(), ApiError> {
    if access.visible(path) == Visible::Nothing {
        return Err(ApiError(
            StatusCode::FORBIDDEN,
            Some("no capability to read this (docs/access.md)".to_owned()),
        ));
    }
    Ok(())
}

/// A read of `path` as `query` asks (the empty path: the whole store), as much of it as
/// `access` lets its user read (`read_shown`). 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,
    access: Access,
) -> Result<Response, ApiError> {
    readable(&access, &path)?;
    let root = path.is_empty();
    if query.shallow {
        return shallow(state, path, root, query, access).await;
    }
    let accept = headers.get(ACCEPT).cloned();
    let want_conflicts = query.conflicts;
    let raw = query.raw;
    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,
                    raw,
                    &access
                ))
            },
            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),
                    signer: None,
                }
                .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>>,
    /// The user who signed it, if it is a value its user signed, opened (`opened`).
    signer: Option<String>,
}

impl Read {
    fn respond(self) -> Result<Response, ApiError> {
        let mut response = if let Some(conflicts) = self.conflicts {
            let value = self.json.unwrap_or(JsonValue::Null);
            let body = serde_json::json!({ "value": value, "conflicts": conflicts });
            show(&self.held, Shown::Json(JsonForm::default()), Some(&body))?
        } else {
            show(&self.held, self.shown, self.json.as_ref())?
        };
        if let Some(signer) = self.signer
            && let Ok(value) = HeaderValue::from_str(&signer)
        {
            response.headers_mut().insert(SIGNED_BY_HEADER, value);
        }
        Ok(response)
    }
}

/// Header naming the user who signed a value read (`access::envelope`).
const SIGNED_BY_HEADER: &str = "x-continuum-signed-by";

/// What a value read is, its signed values opened (`opened`).
enum Opened {
    Plain,
    /// A value its user signed, this user.
    By(String),
    /// A signed `null`: nothing.
    Deleted,
}

/// The values in `json` (at `at`) their users signed (`access::envelope`, at `signed`)
/// opened: each envelope as the value it holds, unchecked (it was checked before it went
/// in), a signed `null` as nothing. What `json` itself is.
fn opened(json: &mut JsonValue, at: &mut Vec<String>, signed: &[Vec<String>]) -> Opened {
    if let JsonValue::String(text) = json
        && envelope::is_signed(signed, at)
    {
        let Ok(payload) = envelope::read(text) else {
            return Opened::Plain;
        };
        if payload.value.is_null() {
            return Opened::Deleted;
        }
        *json = payload.value;
        return Opened::By(payload.user);
    }
    if let JsonValue::Object(entries) = json
        && envelope::above_signed(signed, at)
    {
        entries.retain(|key, value| {
            at.push(key.clone());
            let kept = !matches!(opened(value, at, signed), Opened::Deleted);
            at.pop();
            kept
        });
    }
    Opened::Plain
}

/// What `path` holds, read to show as `accept` asks; `406` if it can't be. Of a place
/// `access` lets its user read only partly, what it may (`access::prune`), as JSON;
/// `403` for a native form, or if nothing of it is. Values their users signed opened
/// (`opened`), unless `raw`, or shown as typed JSON or natively.
fn read_shown(
    doc: &impl crate::document::ReadDoc,
    path: &[String],
    accept: Option<&HeaderValue>,
    conflicts: bool,
    raw: bool,
    access: &Access,
) -> 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,
        ),
    };
    let mut json = json;
    let mut signer = None;
    let openable = match shown {
        Shown::Json(form) => !form.typed,
        Shown::Text => true,
        Shown::Native(_) => false,
    };
    if !raw
        && openable
        && let Some(value) = json.as_mut()
    {
        match opened(value, &mut path.to_vec(), &envelope::signed_paths(doc)) {
            Opened::Plain => {}
            Opened::By(user) => signer = Some(user),
            Opened::Deleted => return Ok(None),
        }
    }
    let (json, conflicts) = match (access.visible(path), shown) {
        (Visible::All, _) => (json, conflicts),
        (_, Shown::Native(_)) => return Err(StatusCode::FORBIDDEN.into()),
        (_, shown) => {
            let typed = matches!(shown, Shown::Json(form) if form.typed);
            let json = match json {
                Some(json) => Some(
                    prune(json, &mut path.to_vec(), access, typed)
                        .ok_or(ApiError(StatusCode::FORBIDDEN, None))?,
                ),
                None => None,
            };
            let conflicts = conflicts.map(|mut conflicts: Vec<store::ConflictReport>| {
                conflicts.retain(|report| access.visible(&report.path) != Visible::Nothing);
                conflicts
            });
            (json, conflicts)
        }
    };
    Ok(Some(Read {
        held,
        shown,
        json,
        conflicts,
        signer,
    }))
}

/// `?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,
    access: Access,
) -> 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)
                    .map(
                        |value| value.map(|value| prune_shallow(value, &path, &access))
                    ))
            },
            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,
    /// Whether the change is signed whatever its path (`document::signatures`).
    #[serde(default)]
    signed: 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>,
    Extension(access): Extension<Access>,
    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());
    }

    // A user's own entry, signed by it: its keys are its own to change, whatever else it
    // may do (`envelope::check` takes nothing else changed).
    let own_entry = media_type(&headers) == SIGNED
        && parse_key_path(&key)
            .is_ok_and(|path| path == crate::access::registry::user_path(&user.0));
    let path = if own_entry {
        crate::access::registry::user_path(&user.0)
    } else {
        parse_write_path(&state, &user, &access, &key, "PUT").await?
    };
    let sign = Sign::Write {
        path: path.clone(),
        asked: query.signed,
    };
    if let Some(redirect) = to_positive(&state, &path, false, None, raw_query.as_deref()).await? {
        return Ok(redirect);
    }
    if media_type(&headers) == SIGNED {
        return put_signed(&state, &user, path, sign, &headers, &body).await;
    }
    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 writer = user.0.clone();
    let ((retyped, read), heads) = state
        .mutate_doc_for(Some(user.0.clone()), sign.clone(), move |doc| {
            unsigned_kept_out(&*doc, &path)?;
            let before = Before::of(&*doc, &writer);
            let retyped = store::write_val_path(doc, &path, &body.value, writing)
                .map_err(|e| store_status(e, "Failed to write store value"))?;
            sealed_kept(&*doc, &path)?;
            registry_kept(&*doc, &path, before, &writer)?;
            // What was written, shown back to whoever wrote it.
            let read = read_shown(
                &*doc,
                &path,
                accept.as_ref(),
                false,
                false,
                &Access::Everything,
            )?;
            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))
}

/// A signed value's media type (`access::envelope`): its payload's JSON, the exact bytes
/// signed, on the first line; then its user's signature of them, armored, as
/// `ssh-keygen -Y sign -n continuum-value` writes it.
const SIGNED: &str = "application/continuum-signed";

/// `PUT` of a value its user signed (`SIGNED`), at a path for one
/// (`envelope::signed_paths`; `409` elsewhere): taken if the authenticated user signed it,
/// for this path, replacing what is there (`409` if it names another: read again and
/// sign anew), with one of its keys, and may write there (`envelope::check`; `403` if
/// not). Written as its envelope, and answered with the value, as `PUT` is.
async fn put_signed(
    state: &AppState,
    user: &AuthenticatedUser,
    path: Vec<String>,
    sign: Sign,
    headers: &HeaderMap,
    body: &Bytes,
) -> Result<Response, ApiError> {
    let bad = |error: String| ApiError(StatusCode::BAD_REQUEST, Some(error));
    let text = std::str::from_utf8(body).map_err(|_| bad("not UTF-8".to_owned()))?;
    let (payload, armored) = text
        .split_once('\n')
        .ok_or_else(|| bad("its payload's line, then its signature".to_owned()))?;
    let sealed = envelope::seal(payload.as_bytes(), armored).map_err(bad)?;
    let signer = envelope::read(&sealed).map_err(bad)?.user;
    if signer != user.0 {
        return Err(ApiError(
            StatusCode::FORBIDDEN,
            Some(format!("signed by {signer}, not by {}", user.0)),
        ));
    }
    let accept = headers.get(ACCEPT).cloned();
    let writer = user.0.clone();
    let (read, heads) = state
        .mutate_doc_for(Some(user.0.clone()), sign, move |doc| {
            if !envelope::is_signed(&envelope::signed_paths(&*doc), &path) {
                return Err(ApiError(
                    StatusCode::CONFLICT,
                    Some(format!(
                        "/{} holds no value its user signs (docs/access.md)",
                        path.join("/")
                    )),
                ));
            }
            let previous = match store::read_path(&*doc, &path)
                .map_err(|e| store_status(e, "Failed to read store path"))?
            {
                Some(JsonValue::String(text)) if envelope::read(&text).is_ok() => Some(text),
                _ => None,
            };
            if envelope::read(&sealed).map_err(bad)?.prev != envelope::digest(previous.as_deref()) {
                return Err(ApiError(
                    StatusCode::CONFLICT,
                    Some("it doesn't replace the value there (prev): read it again".to_owned()),
                ));
            }
            envelope::check(&*doc, &sealed, &path, previous.as_deref())
                .map_err(|error| ApiError(StatusCode::FORBIDDEN, Some(error)))?;
            let before = Before::of(&*doc, &writer);
            // A plain string, whatever was there: what every node checks it as
            // (`access::rules`).
            let envelope = Val::Scalar(ScalarValue::Str(sealed.into()));
            let writing = store::Writing::new(store::Sticky::No, TimestampFormat::default(), true);
            store::write_val_path(doc, &path, &envelope, writing)
                .map_err(|e| store_status(e, "Failed to write store value"))?;
            sealed_kept(&*doc, &path)?;
            registry_kept(&*doc, &path, before, &writer)?;
            read_shown(
                &*doc,
                &path,
                accept.as_ref(),
                false,
                false,
                &Access::Everything,
            )
        })
        .await?;
    // A signed `null`: nothing there now.
    match read {
        Some(read) => Ok(with_heads(read.respond()?, &heads)),
        None => Ok(with_heads(StatusCode::NO_CONTENT.into_response(), &heads)),
    }
}

/// `409` for a write at `path` that leaves a value not sealed as the swarm declares
/// (`access::sealed::kept`).
fn sealed_kept(doc: &impl crate::document::ReadDoc, path: &[String]) -> Result<(), ApiError> {
    crate::access::sealed::kept(doc, path).map_err(|why| ApiError(StatusCode::CONFLICT, Some(why)))
}

/// `409` for a write at `path` that its user didn't sign, under a strict policy
/// (`access::policy`), where it would reach a value its user signs, at, within or above
/// it (`access::envelope`): there, a signed `PUT` only.
fn unsigned_kept_out(doc: &impl crate::document::ReadDoc, path: &[String]) -> Result<(), ApiError> {
    if crate::access::policy::of(doc).strict()
        && envelope::reaches_signed(&envelope::signed_paths(doc), path)
    {
        return Err(ApiError(
            StatusCode::CONFLICT,
            Some(format!(
                "/{} reaches values their users sign (strict policy): PUT each signed, as {SIGNED}",
                path.join("/")
            )),
        ));
    }
    Ok(())
}

/// `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, if `access` lets the user write there (`403` if not). In a swarm
/// with a user registry, by the user's capabilities (`access.md`); in one without, the
/// swarm's data by anyone, protocol data (`_continuum`: remake records, members, forget
/// records, the users) only by users listed in `--protocol-writers`. Protocol data is
/// what every node acts on, so a write there acts on the whole swarm, and each 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; nor is the swarm ID, nor `_continuum/local`, which names
/// what a node holds alone for capabilities and is never stored. A path inside a list or text is
/// judged where the rules judge the write (`inspect::placed`): at the list.
async fn parse_write_path(
    state: &AppState,
    user: &AuthenticatedUser,
    access: &Access,
    key: &str,
    method: &str,
) -> Result<Vec<String>, StatusCode> {
    let path = parse_key_path(key)?;
    let placed = crate::document::inspect::placed(&state.doc.read().await.doc, &path);
    let protocol = meta::is_protocol(&path);
    // `_continuum` itself, epoch metadata, the swarm ID and what is never stored: nobody.
    let fixed = protocol
        && (path.len() < 2
            || path[1] == meta::EPOCHS
            || path[1] == meta::SWARM
            || path[1] == meta::LOCAL);
    let allowed = match access {
        Access::Everything => !protocol || state.protocol_writers.contains(&user.0),
        Access::Caps(caps) => caps.allows(crate::access::Right::Write, &placed),
    };
    if !protocol && allowed {
        return Ok(path);
    }
    if fixed || !allowed {
        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,
}

/// A request's media type (`Content-Type` without parameters, lowercase); JSON if none.
fn media_type(headers: &HeaderMap) -> String {
    let content_type = headers
        .get(CONTENT_TYPE)
        .and_then(|value| value.to_str().ok())
        .unwrap_or("application/json");
    let media = content_type.split(';').next().unwrap_or(content_type);
    media.trim().to_ascii_lowercase()
}

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(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>,
    Extension(access): Extension<Access>,
    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, &access, &key, "POST").await?;
    let sign = Sign::Write {
        path: path.clone(),
        asked: query.signed,
    };
    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 writer = user.0.clone();
        let (value, heads) = state
            .mutate_doc_for(Some(user.0.clone()), sign.clone(), move |doc| {
                unsigned_kept_out(&*doc, &path)?;
                let before = Before::of(&*doc, &writer);
                let value = store::increment_path(doc, &path, by)
                    .map_err(|e| store_status(e, "Failed to increment a counter"))?;
                sealed_kept(&*doc, &path)?;
                registry_kept(&*doc, &path, before, &writer)?;
                Ok::<_, ApiError>(value)
            })
            .await?;
        return Ok(with_heads(
            {
                let counter = 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 writer = user.0.clone();
        let ((retyped, read), heads) = state
            .mutate_doc_for(Some(user.0.clone()), sign.clone(), move |doc| {
                unsigned_kept_out(&*doc, &path)?;
                let before = Before::of(&*doc, &writer);
                let retyped = store::cast_path(doc, &path, to, timestamps)
                    .map_err(|e| store_status(e, "Failed to cast a store value"))?;
                sealed_kept(&*doc, &path)?;
                registry_kept(&*doc, &path, before, &writer)?;
                let read = read_shown(
                    &*doc,
                    &path,
                    accept.as_ref(),
                    false,
                    false,
                    &Access::Everything,
                )?;
                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 writer = user.0.clone();
    let ((), heads) = state
        .mutate_doc_for(Some(user.0.clone()), sign.clone(), move |doc| {
            unsigned_kept_out(&*doc, &path)?;
            let before = Before::of(&*doc, &writer);
            store::insert_val_path(doc, &path, &value.value)
                .map_err(|e| store_status(e, "Failed to insert into a list"))?;
            sealed_kept(&*doc, &path)?;
            registry_kept(&*doc, &path, before, &writer)
        })
        .await?;
    Ok(with_heads(
        (
            StatusCode::CREATED,
            axum::Json(serde_json::json!({ "index": index })),
        )
            .into_response(),
        &heads,
    ))
}

/// `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>,
    Extension(access): Extension<Access>,
    Path(key): Path<String>,
    Query(query): Query<DeleteQuery>,
    RawQuery(raw_query): RawQuery,
) -> Result<Response, ApiError> {
    let path = parse_write_path(&state, &user, &access, &key, "DELETE").await?;
    let sign = Sign::Write {
        path: path.clone(),
        asked: query.signed,
    };

    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);
    }

    // Pruning stops at the highest map the user may write: none above it goes.
    let prune = query.prune.then(|| {
        (1..=path.len())
            .find(|depth| access.allows(crate::access::Right::Write, &path[..*depth]))
            .unwrap_or(path.len())
    });
    let writer = user.0.clone();
    let ((), heads) = state
        .mutate_doc_for(Some(user.0.clone()), sign.clone(), move |doc| {
            unsigned_kept_out(&*doc, &path)?;
            let before = Before::of(&*doc, &writer);
            store::delete_path_pruning(doc, &path, prune)
                .map_err(|e| store_status(e, "Failed to delete store value"))?;
            registry_kept(&*doc, &path, before, &writer)
        })
        .await?;

    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>,
    Extension(access): Extension<Access>,
) -> Result<Response, ApiError> {
    stream(state, String::new(), Vec::new(), access).await
}

/// `GET /stream/{key}`: follows one value.
pub(crate) async fn stream_store_entry(
    State(state): State<AppState>,
    Extension(access): Extension<Access>,
    Path(key): Path<String>,
) -> Result<Response, ApiError> {
    let path = parse_key_path(&key)?;
    stream(state, key, path, access).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; as much of it as
/// `access` lets its user read, as the stream opened (`access::prune`).
async fn stream(
    state: AppState,
    key: String,
    path: Vec<String>,
    access: Access,
) -> Result<Response, ApiError> {
    readable(&access, &path)?;
    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 shown = move |(value, conflicted): (Option<JsonValue>, bool), path: &[String]| {
        let value = value.and_then(|value| prune(value, &mut path.to_vec(), &access, false));
        (value, conflicted)
    };
    let initial = render_traced(&state, path.clone())
        .await
        .map(|read| shown(read, &path))
        .map_err(|e| store_status(e, "Failed to read initial stream value"))?;

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

    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;
            }
        }
        // When the last line went out: a blank one follows after `heartbeat` without
        // another, so proxies in front don't take the stream for dead.
        let mut sent = tokio::time::Instant::now();

        loop {
            let beat = async {
                match sent.checked_add(heartbeat).filter(|_| !heartbeat.is_zero()) {
                    Some(at) => tokio::time::sleep_until(at).await,
                    None => std::future::pending::<()>().await,
                }
            };
            let received = tokio::select! {
                received = changes.recv() => Some(received),
                () = beat => None,
            };
            let Some(received) = received else {
                sent = tokio::time::Instant::now();
                yield Ok(Bytes::from_static(b"\n"));
                continue;
            };
            let relevant = match received {
                // 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) => shown(result, &path),
                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;
                }
            }
            sent = tokio::time::Instant::now();
            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))
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    fn envelope_of(value: JsonValue) -> String {
        use base64::{Engine, engine::general_purpose::URL_SAFE_NO_PAD};
        let payload = json!({"swarm": "s", "path": ["x"], "value": value, "user": "alice", "at": 1, "prev": ""});
        format!("cv1.{}.c2ln", URL_SAFE_NO_PAD.encode(payload.to_string()))
    }

    fn path(text: &str) -> Vec<String> {
        text.split('/').map(str::to_owned).collect()
    }

    /// Envelopes where values are signed open as what they hold, signed `null`s go;
    /// strings elsewhere, and what isn't an envelope, stay as they are.
    #[test]
    fn signed_values_open() {
        let signed = vec![path("users/*"), path("config")];
        let mut read = json!({
            "users": {
                "alice": envelope_of(json!({"caps": {}})),
                "bob": envelope_of(JsonValue::Null),
                "carol": "plain",
            },
            "config": envelope_of(json!(7)),
            "other": envelope_of(json!(8)),
        });
        assert!(matches!(
            opened(&mut read, &mut Vec::new(), &signed),
            Opened::Plain
        ));
        assert_eq!(
            read["users"],
            json!({"alice": {"caps": {}}, "carol": "plain"})
        );
        assert_eq!(read["config"], json!(7));
        assert_eq!(read["other"], json!(envelope_of(json!(8))));

        let mut one = json!(envelope_of(json!("v")));
        assert!(matches!(
            opened(&mut one, &mut path("config"), &signed),
            Opened::By(user) if user == "alice"
        ));
        assert_eq!(one, json!("v"));
        let mut gone = json!(envelope_of(JsonValue::Null));
        assert!(matches!(
            opened(&mut gone, &mut path("users/bob"), &signed),
            Opened::Deleted
        ));
    }
}