diff options
Diffstat (limited to 'src/api/store.rs')
| -rw-r--r-- | src/api/store.rs | 791 |
1 files changed, 691 insertions, 100 deletions
diff --git a/src/api/store.rs b/src/api/store.rs index d0fc807..2b21c83 100644 --- a/src/api/store.rs +++ b/src/api/store.rs @@ -2,27 +2,30 @@ use std::io::Error as IoError; -use automerge::ChangeHash; use axum::{ Extension, body::{Body, Bytes}, - extract::{Path, Query, State}, + extract::{Path, Query, RawQuery, State}, http::{ HeaderMap, HeaderValue, StatusCode, - header::{ACCEPT, CONTENT_TYPE}, + 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}; +use tracing::{debug, error, instrument, warn}; use super::auth::AuthenticatedUser; -use super::{parse_key_path, parse_write_path, response_for_value, store_status}; -use crate::document::store::{self, StoreError}; +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::swarm::sync::paths_related; +use crate::transport::Hex; #[derive(Debug, Default, Deserialize)] pub(crate) struct ReadQuery { @@ -32,6 +35,60 @@ pub(crate) struct ReadQuery { /// 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)] @@ -40,81 +97,163 @@ pub(crate) struct DeleteQuery { prune: bool, } -fn conflicts_response( - value: JsonValue, - conflicts: Vec<store::ConflictReport>, -) -> Result<Response, StatusCode> { - let body = serde_json::json!({ "value": value, "conflicts": conflicts }); - response_for_value(&body, None) // always JSON -} - /// `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, StatusCode> { +) -> 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( - true, + root, query.heads, move |doc| { - if want_conflicts { - store::read_path_with_conflicts(doc, &Vec::new()) - .map(|found| found.map(|(value, conflicts)| (value, Some(conflicts)))) - } else { - store::read_path(doc, &Vec::new()).map(|found| found.map(|value| (value, None))) - } + read_at!(doc, at.as_deref(), |doc| read_shown( + doc, + &path, + accept.as_ref(), + want_conflicts + )) }, - move |result| { - let (value, conflicts) = result - .map_err(|e| store_status(e, "Failed to read store path"))? - .unwrap_or_else(|| (JsonValue::Object(Default::default()), Some(Vec::new()))); - match conflicts { - Some(conflicts) => conflicts_response(value, conflicts), - None => response_for_value(&value, accept.as_ref()), + 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) } -/// `GET /store/{key}`. -pub(crate) async fn get_store_entry( - State(state): State<AppState>, - Path(key): Path<String>, - headers: HeaderMap, - Query(query): Query<ReadQuery>, -) -> Result<Response, StatusCode> { - let path = parse_key_path(&key)?; +/// 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>>, +} - let accept = headers.get(ACCEPT).cloned(); - let want_conflicts = query.conflicts; +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( - false, + unready_live, query.heads, - move |doc| { - if want_conflicts { - store::read_path_with_conflicts(doc, &path) - .map(|found| found.map(|(value, conflicts)| (value, Some(conflicts)))) - } else { - store::read_path(doc, &path).map(|found| found.map(|value| (value, None))) - } - }, + move |doc| read_at!(doc, at.as_deref(), |doc| store::read_shallow(doc, &path)), move |result| { - let (value, conflicts) = result + let value = result .map_err(|e| store_status(e, "Failed to read store path"))? .ok_or(StatusCode::NOT_FOUND)?; - match conflicts { - Some(conflicts) => conflicts_response(value, conflicts), - None => response_for_value(&value, accept.as_ref()), - } + respond(None, "application/json", json_bytes(&value)?) }, ) .await?; @@ -125,24 +264,30 @@ pub(crate) async fn get_store_entry( 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`). +/// 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, StatusCode>, + response: Result<Response, ApiError>, source: ReadSource, heads: Option<Vec<ChangeHash>>, -) -> Result<Response, StatusCode> { +) -> Result<Response, ApiError> { let mut response = response?; if let Some(heads) = heads { - 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 = with_heads(response, &heads); } if let ReadSource::Stale(epoch) = source && let Ok(value) = HeaderValue::from_str(&epoch.to_string()) @@ -152,46 +297,431 @@ fn mark_source( Ok(response) } -/// `PUT /store/{key}`. +#[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, StatusCode> { +) -> Result<Response, ApiError> { if !state.is_ready().await { - return Err(StatusCode::SERVICE_UNAVAILABLE); + 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") - .split(';') - .next() - .unwrap_or("") - .trim(); - - let value = match content_type { - "application/json" => { - serde_json::from_slice::<JsonValue>(&body).map_err(|_| StatusCode::BAD_REQUEST)? - } - "text/plain" => JsonValue::String( - String::from_utf8(body.to_vec()).map_err(|_| StatusCode::BAD_REQUEST)?, + .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, ), - _ => return Err(StatusCode::UNSUPPORTED_MEDIA_TYPE), + 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, + }) +} - let written = value.clone(); - state - .mutate_doc(move |doc| store::write_path(doc, &path, &written)) - .await - .map_err(|e| store_status(e, "Failed to write store value"))?; - - response_for_value(&value, headers.get(ACCEPT)) +/// `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}`. @@ -201,26 +731,89 @@ pub(crate) async fn delete_store_entry( Extension(user): Extension<AuthenticatedUser>, Path(key): Path<String>, Query(query): Query<DeleteQuery>, -) -> Result<StatusCode, StatusCode> { + 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); + return Err(StatusCode::SERVICE_UNAVAILABLE.into()); } if path.is_empty() { - return Err(StatusCode::BAD_REQUEST); + 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; - state - .mutate_doc(move |doc| store::delete_path(doc, &path, 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(StatusCode::NO_CONTENT) + 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 ----------------------------------------------------------------- @@ -239,9 +832,7 @@ async fn render_traced( /// `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, StatusCode> { +pub(crate) async fn stream_store_root(State(state): State<AppState>) -> Result<Response, ApiError> { stream(state, String::new(), Vec::new()).await } @@ -249,16 +840,16 @@ pub(crate) async fn stream_store_root( pub(crate) async fn stream_store_entry( State(state): State<AppState>, Path(key): Path<String>, -) -> Result<Response, StatusCode> { +) -> 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, StatusCode> { +async fn stream(state: AppState, key: String, path: Vec<String>) -> Result<Response, ApiError> { if !state.is_ready().await { - return Err(StatusCode::SERVICE_UNAVAILABLE); + return Err(StatusCode::SERVICE_UNAVAILABLE.into()); } // Subscribe first, so nothing between the initial read and the loop is missed. |