aboutsummaryrefslogtreecommitdiffstats
path: root/src/api/store.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/api/store.rs')
-rw-r--r--src/api/store.rs364
1 files changed, 287 insertions, 77 deletions
diff --git a/src/api/store.rs b/src/api/store.rs
index 1478d14..13f5012 100644
--- a/src/api/store.rs
+++ b/src/api/store.rs
@@ -8,7 +8,7 @@ use axum::{
extract::{Path, Query, RawQuery, State},
http::{
HeaderMap, HeaderValue, StatusCode,
- header::{ACCEPT, CONTENT_TYPE, ETAG, LOCATION},
+ header::{ACCEPT, CONTENT_TYPE, ETAG, LOCATION, VARY},
},
response::{IntoResponse, Response},
};
@@ -19,10 +19,11 @@ use tokio::sync::broadcast;
use tracing::{debug, error, instrument, warn};
use super::auth::AuthenticatedUser;
-use super::{ApiError, media_type, store_status};
+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};
use crate::state::{AppState, ReadSource};
use crate::transport::Hex;
@@ -71,7 +72,7 @@ macro_rules! read_at {
let $d = &at;
$read
}
- None => Err(StoreError::UnknownHeads),
+ None => Err(From::from(StoreError::UnknownHeads)),
},
}
};
@@ -90,14 +91,6 @@ pub(crate) struct DeleteQuery {
prune: bool,
}
-fn conflicts_response(
- value: JsonValue,
- conflicts: Vec<store::ConflictReport>,
-) -> Result<Response, ApiError> {
- 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>,
@@ -147,30 +140,92 @@ async fn read(
root,
query.heads,
move |doc| {
- read_at!(doc, at.as_deref(), |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)))
- })
+ read_at!(doc, at.as_deref(), |doc| read_shown(
+ doc,
+ &path,
+ accept.as_ref(),
+ want_conflicts
+ ))
},
- move |result| {
- let found = result.map_err(|e| store_status(e, "Failed to read store path"))?;
- let (value, conflicts) = match found {
- Some(found) => found,
- None if root => (JsonValue::Object(Default::default()), Some(Vec::new())),
- None => return Err(StatusCode::NOT_FOUND.into()),
- };
- 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)
}
+/// 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(
@@ -192,7 +247,7 @@ async fn shallow(
let value = result
.map_err(|e| store_status(e, "Failed to read store path"))?
.ok_or(StatusCode::NOT_FOUND)?;
- response_for_value(&value, None)
+ respond(None, "application/json", json_bytes(&value)?)
},
)
.await?;
@@ -254,19 +309,19 @@ pub(crate) async fn put_store_entry(
return Ok(redirect);
}
let value = body_value(&headers, &body)?;
+ // Answered with the value written, shown as asked.
+ let (held, json) = value.held();
+ let shown = shown(headers.get(ACCEPT), held.kind)?;
- let written = value.clone();
let ((), heads) = state
- .mutate_doc_for(Some(user.0.clone()), move |doc| {
- store::write_path(doc, &path, &written)
+ .mutate_doc_for(Some(user.0.clone()), move |doc| match &value {
+ Written::Json(value) => store::write_path(doc, &path, value),
+ Written::Scalar(scalar) => store::write_scalar_path(doc, &path, scalar),
})
.await
.map_err(|e| store_status(e, "Failed to write store value"))?;
- Ok(with_heads(
- response_for_value(&value, headers.get(ACCEPT))?,
- &heads,
- ))
+ Ok(with_heads(show(&held, shown, json.as_ref())?, &heads))
}
fn etag_for(bytes: &[u8]) -> String {
@@ -309,62 +364,205 @@ fn parse_write_path(
Ok(path)
}
-fn response_for_value(
- value: &JsonValue,
- accept: Option<&HeaderValue>,
-) -> Result<Response, ApiError> {
- let wants_text = accept
- .and_then(|value| value.to_str().ok())
- .is_some_and(|value| {
- value
- .split(',')
- .any(|item| media_type(item) == "text/plain")
- });
+/// 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),
+}
- let (content_type, body) = if wants_text {
- let body = match value {
- JsonValue::String(value) => value.clone(),
- other => serde_json::to_string(other).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?,
- };
+/// 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) {
+ if media == "application/json" {
+ return json_form_of(params);
+ }
+ }
+ Ok(JsonForm::default())
+}
- ("text/plain; charset=utf-8", body)
- } else {
- (
- "application/json",
- serde_json::to_string(value).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?,
- )
+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 })
+}
- let etag = etag_for(body.as_bytes());
+/// 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()
+}
- let mut response = body.into_response();
+/// 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::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::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(", ")
+ )),
+ ))
+}
- response
- .headers_mut()
- .insert(CONTENT_TYPE, HeaderValue::from_static(content_type));
+/// 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(_) => respond(Some(held.kind), "application/json", 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)
+ }
+ }
+}
- response.headers_mut().insert(
+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)
}
-/// A write's body as a value: JSON (the default), or text as a string.
-fn body_value(headers: &HeaderMap, body: &Bytes) -> Result<JsonValue, StatusCode> {
+/// 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: JSON (the default), text as a string, or a value of a type of its
+/// own (`typed::from_native`).
+enum Written {
+ Json(JsonValue),
+ Scalar(automerge::ScalarValue),
+}
+
+impl Written {
+ /// What it is stored as, and its JSON, to answer the write with.
+ fn held(&self) -> (store::Held, Option<JsonValue>) {
+ match self {
+ Written::Json(json) => {
+ let kind = match json {
+ JsonValue::Object(_) => Kind::Map,
+ JsonValue::Array(_) => Kind::List,
+ JsonValue::String(_) => Kind::String,
+ JsonValue::Bool(_) => Kind::Boolean,
+ JsonValue::Null => Kind::Null,
+ JsonValue::Number(number) if number.is_i64() => Kind::Int,
+ JsonValue::Number(number) if number.is_u64() => Kind::Uint,
+ JsonValue::Number(_) => Kind::Float,
+ };
+ let value = store::json_scalar(json).ok().flatten();
+ (store::Held { kind, value }, Some(json.clone()))
+ }
+ Written::Scalar(scalar) => (
+ store::Held {
+ kind: Kind::of(scalar),
+ value: Some(scalar.clone()),
+ },
+ typed::scalar_to_json(scalar, JsonForm::default()),
+ ),
+ }
+ }
+}
+
+fn body_value(headers: &HeaderMap, body: &Bytes) -> Result<Written, ApiError> {
let content_type = headers
.get(CONTENT_TYPE)
.and_then(|value| value.to_str().ok())
.unwrap_or("application/json");
- match media_type(content_type) {
- "application/json" => {
- serde_json::from_slice::<JsonValue>(body).map_err(|_| StatusCode::BAD_REQUEST)
- }
- "text/plain" => String::from_utf8(body.to_vec())
- .map(JsonValue::String)
- .map_err(|_| StatusCode::BAD_REQUEST),
- _ => Err(StatusCode::UNSUPPORTED_MEDIA_TYPE),
+ let (media, params) = content_type.split_once(';').unwrap_or((content_type, ""));
+ let media = media.trim().to_ascii_lowercase();
+ match media.as_str() {
+ "application/json" => serde_json::from_slice::<JsonValue>(body)
+ .map(Written::Json)
+ .map_err(|_| StatusCode::BAD_REQUEST.into()),
+ typed::TEXT => String::from_utf8(body.to_vec())
+ .map(|text| Written::Json(JsonValue::String(text)))
+ .map_err(|_| StatusCode::BAD_REQUEST.into()),
+ media => match typed::from_native(media, params, body) {
+ Some(Ok(scalar)) => Ok(Written::Scalar(scalar)),
+ Some(Err(error)) => Err(ApiError(StatusCode::BAD_REQUEST, Some(error))),
+ None => Err(StatusCode::UNSUPPORTED_MEDIA_TYPE.into()),
+ },
}
}
@@ -402,15 +600,27 @@ pub(crate) async fn post_store_entry(
.await
.map_err(|e| store_status(e, "Failed to increment a counter"))?;
return Ok(with_heads(
- response_for_value(&JsonValue::from(value), headers.get(ACCEPT))?,
+ {
+ 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,
));
}
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_path(doc, &path, &value)
+ .mutate_doc_for(Some(user.0.clone()), move |doc| match &value {
+ Written::Json(value) => store::insert_path(doc, &path, value),
+ Written::Scalar(scalar) => store::insert_scalar_path(doc, &path, scalar),
})
.await;
match inserted {