diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/api/mod.rs | 7 | ||||
| -rw-r--r-- | src/api/store.rs | 364 | ||||
| -rw-r--r-- | src/document/mod.rs | 1 | ||||
| -rw-r--r-- | src/document/store.rs | 200 | ||||
| -rw-r--r-- | src/document/typed.rs | 318 |
5 files changed, 768 insertions, 122 deletions
diff --git a/src/api/mod.rs b/src/api/mod.rs index dfefcc8..713f248 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -199,6 +199,13 @@ impl IntoResponse for ApiError { /// A store error as an answer, saying why (`StoreError`'s text) where it is the /// client's; others are logged with `context`, and are a bare `500`. +/// A read's (`store_status`). +impl From<StoreError> for ApiError { + fn from(error: StoreError) -> Self { + store_status(error, "Failed to read store path") + } +} + pub(crate) fn store_status(error: StoreError, context: &str) -> ApiError { let status = match error { StoreError::Conflict 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 { diff --git a/src/document/mod.rs b/src/document/mod.rs index d7c2fcb..4d7fbfe 100644 --- a/src/document/mod.rs +++ b/src/document/mod.rs @@ -46,6 +46,7 @@ pub(crate) mod store; pub(crate) mod sync; #[cfg(test)] pub(crate) mod testing; +pub(crate) mod typed; #[cfg(test)] mod tests { diff --git a/src/document/store.rs b/src/document/store.rs index cc67f77..0f1fd58 100644 --- a/src/document/store.rs +++ b/src/document/store.rs @@ -5,11 +5,12 @@ use automerge::{ ActorId, Automerge, ObjId, ObjType, Prop, ReadDoc, ScalarValue, Value, transaction::Transactable, }; -use base64::{Engine, engine::general_purpose::STANDARD as BASE64}; use serde::Serialize; use serde_json::Value as JsonValue; use std::collections::BTreeSet; +use super::typed::{self, JsonForm, Kind}; + /// Store paths touched by one batch of document changes (`AppState::change_events`). #[derive(Debug)] pub(crate) struct ChangeSet { @@ -283,14 +284,15 @@ fn merged_keys(doc: &impl ReadDoc, maps: &[ObjId]) -> BTreeSet<String> { fn resolved_to_json( doc: &impl ReadDoc, resolved: &Resolved, + form: JsonForm, mut log: Option<&mut ConflictLog>, ) -> Result<JsonValue, StoreError> { let maps = match resolved { Resolved::Maps(maps) => maps, Resolved::Other(Value::Scalar(scalar), _) => { - return scalar_to_json(scalar.clone().into_owned()); + return scalar_to_json(scalar.clone().into_owned(), form); } - Resolved::Other(Value::Object(_), id) => return object_to_json(doc, id, log), + Resolved::Other(Value::Object(_), id) => return object_to_json(doc, id, form, log), }; let mut out = serde_json::Map::new(); @@ -313,7 +315,9 @@ fn resolved_to_json( } }; - let json = nested(&mut log, &key, |log| resolved_to_json(doc, &child, log))?; + let json = nested(&mut log, &key, |log| { + resolved_to_json(doc, &child, form, log) + })?; out.insert(key, json); } } else { @@ -322,7 +326,9 @@ fn resolved_to_json( let Some(child) = resolve(doc, maps, &key, log.as_deref_mut())? else { continue; }; - let json = nested(&mut log, &key, |log| resolved_to_json(doc, &child, log))?; + let json = nested(&mut log, &key, |log| { + resolved_to_json(doc, &child, form, log) + })?; out.insert(key, json); } } @@ -333,19 +339,20 @@ fn resolved_to_json( fn object_to_json( doc: &impl ReadDoc, id: &ObjId, + form: JsonForm, mut log: Option<&mut ConflictLog>, ) -> Result<JsonValue, StoreError> { match doc.object_type(id)? { - ObjType::Map => resolved_to_json(doc, &Resolved::Maps(vec![id.clone()]), log), + ObjType::Map => resolved_to_json(doc, &Resolved::Maps(vec![id.clone()]), form, log), ObjType::List => { let mut out = Vec::new(); for (index, item) in doc.list_range(id, ..).enumerate() { let child_id = item.id(); out.push(match Value::from(item.value) { Value::Object(_) => nested(&mut log, &index.to_string(), |log| { - object_to_json(doc, &child_id, log) + object_to_json(doc, &child_id, form, log) })?, - Value::Scalar(scalar) => scalar_to_json(scalar.into_owned())?, + Value::Scalar(scalar) => scalar_to_json(scalar.into_owned(), form)?, }); } Ok(JsonValue::Array(out)) @@ -367,12 +374,13 @@ fn delete_in_all(doc: &mut impl Transactable, maps: &[ObjId], key: &str) -> Resu fn read_path_inner( doc: &impl ReadDoc, path: &[String], + form: JsonForm, mut log: Option<&mut ConflictLog>, ) -> Result<Option<JsonValue>, StoreError> { let Some(current) = resolve_logged(doc, path, log.as_deref_mut())? else { return Ok(None); }; - Ok(Some(resolved_to_json(doc, ¤t, log)?)) + Ok(Some(resolved_to_json(doc, ¤t, form, log)?)) } /// Reads a path; the empty path is the whole store. @@ -380,21 +388,82 @@ pub(crate) fn read_path( doc: &impl ReadDoc, path: &[String], ) -> Result<Option<JsonValue>, StoreError> { - read_path_inner(doc, path, None) + read_path_as(doc, path, JsonForm::default()) +} + +/// [`read_path`], in JSON of `form`. +pub(crate) fn read_path_as( + doc: &impl ReadDoc, + path: &[String], + form: JsonForm, +) -> Result<Option<JsonValue>, StoreError> { + read_path_inner(doc, path, form, None) } +#[cfg(test)] pub(crate) fn read_path_with_conflicts( doc: &impl ReadDoc, path: &[String], ) -> Result<Option<(JsonValue, Vec<ConflictReport>)>, StoreError> { + read_path_with_conflicts_as(doc, path, JsonForm::default()) +} + +/// [`read_path_with_conflicts`], in JSON of `form`. +pub(crate) fn read_path_with_conflicts_as( + doc: &impl ReadDoc, + path: &[String], + form: JsonForm, +) -> Result<Option<(JsonValue, Vec<ConflictReport>)>, StoreError> { let mut log = ConflictLog { want_values: true, + form, ..Default::default() }; - let value = read_path_inner(doc, path, Some(&mut log))?; + let value = read_path_inner(doc, path, form, Some(&mut log))?; Ok(value.map(|value| (value, log.reports))) } +/// What is stored at `path` (the empty path: the root map), and, where it is a scalar +/// or a text, the value itself, to show natively (`typed::native`); None if nothing is +/// there. +pub(crate) fn read_held(doc: &impl ReadDoc, path: &[String]) -> Result<Option<Held>, StoreError> { + let Some(resolved) = resolve_through(doc, path)? else { + return Ok(None); + }; + Ok(Some(match resolved { + Resolved::Maps(_) | Resolved::Other(Value::Object(ObjType::Map), _) => Held { + kind: Kind::Map, + value: None, + }, + Resolved::Other(Value::Object(ObjType::List), _) => Held { + kind: Kind::List, + value: None, + }, + Resolved::Other(Value::Object(ObjType::Text), id) => Held { + kind: Kind::Text, + value: Some(ScalarValue::Str(doc.text(&id)?.into())), + }, + Resolved::Other(Value::Object(ObjType::Table), _) => { + return Err(anyhow!("Automerge tables are not supported").into()); + } + Resolved::Other(Value::Scalar(scalar), _) => { + let scalar = scalar.into_owned(); + Held { + kind: Kind::of(&scalar), + value: Some(scalar), + } + } + })) +} + +/// What a path holds (`read_held`). +#[derive(Debug, Clone)] +pub(crate) struct Held { + pub(crate) kind: Kind, + /// A scalar, or a text's characters as a string. + pub(crate) value: Option<ScalarValue>, +} + /// For streams: the value, plus whether the path or subtree involves any conflict. /// Reports are collected without values, so this costs little more than read_path. pub(crate) fn read_path_traced( @@ -402,7 +471,7 @@ pub(crate) fn read_path_traced( path: &[String], ) -> Result<(Option<JsonValue>, bool), StoreError> { let mut log = ConflictLog::default(); - let value = read_path_inner(doc, path, Some(&mut log))?; + let value = read_path_inner(doc, path, JsonForm::default(), Some(&mut log))?; Ok((value, !log.reports.is_empty())) } @@ -506,20 +575,21 @@ fn summary( return Err(anyhow!("Automerge tables are not supported").into()); } Resolved::Other(Value::Scalar(scalar), _) => { - let (kind, size) = match &**scalar { - ScalarValue::Str(text) => ("string", Some(text.len())), - ScalarValue::Bytes(bytes) => ("bytes", Some(bytes.len())), - ScalarValue::Null => ("null", None), - ScalarValue::Boolean(_) => ("boolean", None), - ScalarValue::Unknown { .. } => ("unknown", None), - _ => ("number", None), + let kind = Kind::of(scalar); + let size = match &**scalar { + ScalarValue::Str(text) => Some(text.len()), + ScalarValue::Bytes(bytes) => Some(bytes.len()), + _ => None, }; - out.insert("type".into(), kind.into()); + out.insert("type".into(), kind.name().into()); if let Some(size) = size { out.insert("size".into(), size.into()); } - if kind != "unknown" && size.is_none_or(|size| size <= SHALLOW_INLINE) { - out.insert("value".into(), scalar_to_json(scalar.clone().into_owned())?); + if kind != Kind::Unknown && size.is_none_or(|size| size <= SHALLOW_INLINE) { + out.insert( + "value".into(), + scalar_to_json(scalar.clone().into_owned(), JsonForm::default())?, + ); } } } @@ -547,8 +617,8 @@ pub(crate) fn read_candidates( candidates(doc, &maps, leaf.as_str())? .into_iter() .map(|(value, id)| match value { - Value::Scalar(scalar) => scalar_to_json(scalar.into_owned()), - Value::Object(_) => object_to_json(doc, &id, None), + Value::Scalar(scalar) => scalar_to_json(scalar.into_owned(), JsonForm::default()), + Value::Object(_) => object_to_json(doc, &id, JsonForm::default(), None), }) .collect() } @@ -581,6 +651,38 @@ pub(crate) fn write_path( Ok(()) } +/// [`write_path`] for a scalar (`typed::from_native`): one of a type JSON doesn't have, +/// or doesn't tell apart (bytes, a timestamp, an unsigned integer, a counter). +pub(crate) fn write_scalar_path( + doc: &mut impl Transactable, + path: &[String], + scalar: &ScalarValue, +) -> Result<(), StoreError> { + let (leaf, parents) = path.split_last().ok_or(StoreError::NotFound)?; + let same = |holds: &[(Value<'static>, ObjId)]| matches!(holds, [(Value::Scalar(held), _)] if held.as_ref() == scalar); + match parent_for_write(doc, parents)? { + Parent::Maps(maps) => { + if !same(&candidates(doc, &maps, leaf.as_str())?) { + delete_in_all(doc, &maps, leaf)?; + put_scalar(doc, &maps[0], Slot::Key(leaf), scalar.clone())?; + } + } + Parent::List(list) => { + let index = list_index(leaf).ok_or(StoreError::NotFound)?; + let length = doc.length(&list); + if index > length { + return Err(StoreError::NotFound); + } + if index == length { + put_scalar(doc, &list, Slot::Index(index), scalar.clone())?; + } else if !same(&candidates(doc, std::slice::from_ref(&list), index)?) { + put_scalar(doc, &list, Slot::At(index), scalar.clone())?; + } + } + } + Ok(()) +} + // A write makes its path read as the value written, changing only what differs: a map // written over a map keeps its keys that stay (and what is inside them), and gets // those it lacks and loses those the value doesn't have; a list written over a list @@ -591,7 +693,7 @@ pub(crate) fn write_path( // conflict. /// What `value` would be stored as, if it is a scalar. -fn json_scalar(value: &JsonValue) -> Result<Option<ScalarValue>, StoreError> { +pub(crate) fn json_scalar(value: &JsonValue) -> Result<Option<ScalarValue>, StoreError> { Ok(Some(match value { JsonValue::String(value) => ScalarValue::Str(value.as_str().into()), JsonValue::Bool(value) => ScalarValue::Boolean(*value), @@ -815,6 +917,22 @@ pub(crate) fn insert_path( path: &[String], value: &JsonValue, ) -> Result<(), StoreError> { + let (list, index) = insertion(doc, path)?; + write_value(doc, &list, Slot::Index(index), value) +} + +/// [`insert_path`] for a scalar (`typed::from_native`). +pub(crate) fn insert_scalar_path( + doc: &mut impl Transactable, + path: &[String], + scalar: &ScalarValue, +) -> Result<(), StoreError> { + let (list, index) = insertion(doc, path)?; + put_scalar(doc, &list, Slot::Index(index), scalar.clone()) +} + +/// The list an insert at `path` goes into, and where (`insert_path`). +fn insertion(doc: &impl ReadDoc, path: &[String]) -> Result<(ObjId, usize), StoreError> { let (leaf, parents) = path.split_last().ok_or(StoreError::NotFound)?; let list = match resolve_through(doc, parents)? { None => return Err(StoreError::NotFound), @@ -826,8 +944,7 @@ pub(crate) fn insert_path( if index > length { return Err(StoreError::PastTheEnd { index, length }); } - write_value(doc, &list, Slot::Index(index), value)?; - Ok(()) + Ok((list, index)) } /// Makes `dst` hold at `path` what `src`'s merged view holds there: a type-preserving @@ -1150,21 +1267,11 @@ fn json_to_automerge( write_value(doc, parent, Slot::Key(key), value) } -fn scalar_to_json(scalar: ScalarValue) -> Result<JsonValue, StoreError> { - match scalar { - ScalarValue::Null => Ok(JsonValue::Null), - ScalarValue::Boolean(b) => Ok(JsonValue::Bool(b)), - ScalarValue::Int(i) => Ok(JsonValue::Number(i.into())), - ScalarValue::Uint(u) => Ok(JsonValue::Number(u.into())), - ScalarValue::F64(f) => serde_json::Number::from_f64(f) - .map(JsonValue::Number) - .ok_or(anyhow!("Non-finite float cannot be represented as JSON").into()), - ScalarValue::Str(s) => Ok(JsonValue::String(s.to_string())), - ScalarValue::Bytes(b) => Ok(JsonValue::String(BASE64.encode(b))), - ScalarValue::Counter(c) => Ok(JsonValue::Number(i64::from(c).into())), - ScalarValue::Timestamp(t) => Ok(JsonValue::Number(t.into())), - ScalarValue::Unknown { .. } => Err(anyhow!("Unsupported Automerge scalar type").into()), - } +fn scalar_to_json(scalar: ScalarValue, form: JsonForm) -> Result<JsonValue, StoreError> { + typed::scalar_to_json(&scalar, form).ok_or_else(|| match scalar { + ScalarValue::F64(_) => anyhow!("Non-finite float cannot be represented as JSON").into(), + _ => anyhow!("Unsupported Automerge scalar type").into(), + }) } // --- conflict log ------------------------------------------------------------ @@ -1204,6 +1311,8 @@ struct ConflictLog { want_values: bool, /// Currently walking the ancestors of the requested path. in_ancestors: bool, + /// How candidate values are shown. + form: JsonForm, } fn op_id_string(id: &ObjId) -> String { @@ -1217,10 +1326,11 @@ fn value_to_json( doc: &impl ReadDoc, value: &Value<'_>, id: &ObjId, + form: JsonForm, ) -> Result<JsonValue, StoreError> { match value { - Value::Scalar(scalar) => Ok(scalar_to_json(scalar.clone().into_owned())?), - Value::Object(_) => object_to_json(doc, id, None), + Value::Scalar(scalar) => Ok(scalar_to_json(scalar.clone().into_owned(), form)?), + Value::Object(_) => object_to_json(doc, id, form, None), } } @@ -1247,7 +1357,7 @@ impl ConflictLog { op_id: op_id_string(id), selected, value: if with_values { - Some(value_to_json(doc, value, id)?) + Some(value_to_json(doc, value, id, self.form)?) } else { None }, diff --git a/src/document/typed.rs b/src/document/typed.rs new file mode 100644 index 0000000..bca79b3 --- /dev/null +++ b/src/document/typed.rs @@ -0,0 +1,318 @@ +//! The types a value is stored as, and how each is shown: as JSON (every type has a +//! JSON form), and natively, as a media type of its own, where it is a scalar or a text +//! (`docs/api.md`, Types). + +use base64::{Engine, engine::general_purpose::STANDARD as BASE64}; +use serde_json::Value as JsonValue; + +use automerge::ScalarValue; + +/// What a value is stored as. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum Kind { + Map, + List, + /// An Automerge text (characters that merge one by one), shown as a string. + Text, + String, + Bytes, + Int, + Uint, + Float, + Boolean, + Null, + Counter, + Timestamp, + /// A scalar of a later Automerge version. + Unknown, +} + +pub(crate) const OCTET_STREAM: &str = "application/octet-stream"; +pub(crate) const TEXT: &str = "text/plain"; +pub(crate) const INT: &str = "application/vnd.continuum.int"; +pub(crate) const UINT: &str = "application/vnd.continuum.uint"; +pub(crate) const FLOAT: &str = "application/vnd.continuum.float"; +pub(crate) const BOOLEAN: &str = "application/vnd.continuum.boolean"; +pub(crate) const COUNTER: &str = "application/vnd.continuum.counter"; +pub(crate) const TIMESTAMP: &str = "application/vnd.continuum.timestamp"; + +impl Kind { + pub(crate) fn of(scalar: &ScalarValue) -> Self { + match scalar { + ScalarValue::Str(_) => Kind::String, + ScalarValue::Bytes(_) => Kind::Bytes, + ScalarValue::Int(_) => Kind::Int, + ScalarValue::Uint(_) => Kind::Uint, + ScalarValue::F64(_) => Kind::Float, + ScalarValue::Boolean(_) => Kind::Boolean, + ScalarValue::Null => Kind::Null, + ScalarValue::Counter(_) => Kind::Counter, + ScalarValue::Timestamp(_) => Kind::Timestamp, + ScalarValue::Unknown { .. } => Kind::Unknown, + } + } + + /// As `x-continuum-type` and shallow reads name it. + pub(crate) fn name(self) -> &'static str { + match self { + Kind::Map => "map", + Kind::List => "list", + Kind::Text => "text", + Kind::String => "string", + Kind::Bytes => "bytes", + Kind::Int => "int", + Kind::Uint => "uint", + Kind::Float => "float", + Kind::Boolean => "boolean", + Kind::Null => "null", + Kind::Counter => "counter", + Kind::Timestamp => "timestamp", + Kind::Unknown => "unknown", + } + } + + /// The media type a value of this kind is shown as natively, if any. + pub(crate) fn media_type(self) -> Option<&'static str> { + Some(match self { + Kind::Text | Kind::String => TEXT, + Kind::Bytes => OCTET_STREAM, + Kind::Int => INT, + Kind::Uint => UINT, + Kind::Float => FLOAT, + Kind::Boolean => BOOLEAN, + Kind::Counter => COUNTER, + Kind::Timestamp => TIMESTAMP, + Kind::Map | Kind::List | Kind::Null | Kind::Unknown => return None, + }) + } +} + +/// How a timestamp is written out: as text natively (`format=`), and in JSON +/// (`timestamp-format=`). +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub(crate) enum TimestampFormat { + /// `2026-11-01T21:50:20.345Z` (a string in JSON). + #[default] + Rfc3339, + /// Seconds since 1970, with milliseconds as a fraction (a number in JSON). + Unix, + /// Milliseconds since 1970 (an integer in JSON). + UnixMs, + /// `Sun, 01 Nov 2026 21:50:20 GMT` (RFC 9110; whole seconds; a string in JSON). + HttpDate, +} + +impl TimestampFormat { + pub(crate) fn named(name: &str) -> Option<Self> { + Some(match name { + "rfc3339" => Self::Rfc3339, + "unix" => Self::Unix, + "unix-ms" => Self::UnixMs, + "http-date" => Self::HttpDate, + _ => return None, + }) + } + + pub(crate) fn name(self) -> &'static str { + match self { + Self::Rfc3339 => "rfc3339", + Self::Unix => "unix", + Self::UnixMs => "unix-ms", + Self::HttpDate => "http-date", + } + } + + /// `ms` (milliseconds since 1970) as text. + pub(crate) fn text(self, ms: i64) -> String { + match self { + Self::UnixMs => ms.to_string(), + Self::Unix => { + let sign = if ms < 0 { "-" } else { "" }; + let (seconds, millis) = (ms.unsigned_abs() / 1000, ms.unsigned_abs() % 1000); + if millis == 0 { + format!("{sign}{seconds}") + } else { + format!("{sign}{seconds}.{millis:03}") + .trim_end_matches('0') + .to_owned() + } + } + Self::Rfc3339 | Self::HttpDate => match jiff::Timestamp::from_millisecond(ms) { + Ok(at) if self == Self::Rfc3339 => at.to_string(), + Ok(at) => jiff::fmt::rfc2822::DateTimePrinter::new() + .timestamp_to_rfc9110_string(&at) + .unwrap_or_else(|_| ms.to_string()), + // Beyond what a date shows: the number itself. + Err(_) => ms.to_string(), + }, + } + } + + /// `ms` in JSON: a string, or for the Unix formats a number. + pub(crate) fn json(self, ms: i64) -> JsonValue { + match self { + Self::UnixMs => JsonValue::from(ms), + Self::Unix => serde_json::Number::from_f64(ms as f64 / 1000.0) + .map_or(JsonValue::from(ms / 1000), JsonValue::Number), + Self::Rfc3339 | Self::HttpDate => JsonValue::String(self.text(ms)), + } + } + + /// Milliseconds since 1970 from `text` in this format; None if it isn't one, or is + /// finer than a millisecond. + pub(crate) fn read(self, text: &str) -> Option<i64> { + let text = text.trim(); + match self { + Self::UnixMs => text.parse().ok(), + Self::Unix => { + let (seconds, fraction) = text.split_once('.').unwrap_or((text, "")); + if fraction.len() > 3 || !fraction.bytes().all(|byte| byte.is_ascii_digit()) { + return None; + } + let seconds: i64 = seconds.parse().ok()?; + let millis: i64 = format!("{fraction:0<3}").parse().ok()?; + let millis = if text.starts_with('-') { + -millis + } else { + millis + }; + seconds.checked_mul(1000)?.checked_add(millis) + } + Self::Rfc3339 => { + let at: jiff::Timestamp = text.parse().ok()?; + (at.subsec_nanosecond() % 1_000_000 == 0).then(|| at.as_millisecond()) + } + Self::HttpDate => jiff::fmt::rfc2822::DateTimeParser::new() + .parse_timestamp(text) + .ok() + .map(|at| at.as_millisecond()), + } + } +} + +/// How values are shown in JSON. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub(crate) struct JsonForm { + pub(crate) timestamps: TimestampFormat, +} + +/// `scalar` in JSON: integers, floats and counters as numbers, bytes as base64, +/// timestamps as `form` says. +pub(crate) fn scalar_to_json(scalar: &ScalarValue, form: JsonForm) -> Option<JsonValue> { + Some(match scalar { + ScalarValue::Null => JsonValue::Null, + ScalarValue::Boolean(b) => JsonValue::Bool(*b), + ScalarValue::Int(i) => JsonValue::Number((*i).into()), + ScalarValue::Uint(u) => JsonValue::Number((*u).into()), + ScalarValue::F64(f) => JsonValue::Number(serde_json::Number::from_f64(*f)?), + ScalarValue::Str(s) => JsonValue::String(s.to_string()), + ScalarValue::Bytes(b) => JsonValue::String(BASE64.encode(b)), + ScalarValue::Counter(c) => JsonValue::Number(i64::from(c).into()), + ScalarValue::Timestamp(t) => form.timestamps.json(*t), + ScalarValue::Unknown { .. } => return None, + }) +} + +/// `scalar` shown natively: its media type (with the timestamp format) and body; None +/// for a null or an unknown scalar, which have none. +pub(crate) fn native( + scalar: &ScalarValue, + timestamps: TimestampFormat, +) -> Option<(String, Vec<u8>)> { + let media = Kind::of(scalar).media_type()?; + let body = match scalar { + ScalarValue::Str(text) => { + return Some((format!("{TEXT}; charset=utf-8"), text.as_bytes().to_vec())); + } + ScalarValue::Bytes(bytes) => bytes.clone(), + ScalarValue::Int(i) => i.to_string().into_bytes(), + ScalarValue::Uint(u) => u.to_string().into_bytes(), + ScalarValue::F64(f) => f.to_string().into_bytes(), + ScalarValue::Boolean(b) => b.to_string().into_bytes(), + ScalarValue::Counter(c) => i64::from(c).to_string().into_bytes(), + ScalarValue::Timestamp(t) => { + return Some(( + format!("{TIMESTAMP}; format={}", timestamps.name()), + timestamps.text(*t).into_bytes(), + )); + } + ScalarValue::Null | ScalarValue::Unknown { .. } => return None, + }; + Some((media.to_owned(), body)) +} + +/// The value of media type `media` (with its `params`) that `body` holds, for a write; +/// None if `media` isn't one of the native types (`text/plain` is a JSON string's, and +/// read as one by the caller). An error says what's wrong with the body. +pub(crate) fn from_native( + media: &str, + params: &str, + body: &[u8], +) -> Option<Result<ScalarValue, String>> { + let text = || { + std::str::from_utf8(body) + .map(str::trim) + .map_err(|_| "not UTF-8".to_owned()) + }; + Some(match media { + OCTET_STREAM => Ok(ScalarValue::Bytes(body.to_vec())), + INT => text().and_then(|text| { + text.parse() + .map(ScalarValue::Int) + .map_err(|_| format!("not an integer: {text}")) + }), + UINT => text().and_then(|text| { + text.parse() + .map(ScalarValue::Uint) + .map_err(|_| format!("not an unsigned integer: {text}")) + }), + FLOAT => text().and_then(|text| { + text.parse::<f64>() + .ok() + .filter(|f| f.is_finite()) + .map(ScalarValue::F64) + .ok_or_else(|| format!("not a finite number: {text}")) + }), + BOOLEAN => text().and_then(|text| match text { + "true" => Ok(ScalarValue::Boolean(true)), + "false" => Ok(ScalarValue::Boolean(false)), + other => Err(format!("not true or false: {other}")), + }), + COUNTER => text().and_then(|text| { + text.parse::<i64>() + .map(ScalarValue::counter) + .map_err(|_| format!("not an integer: {text}")) + }), + TIMESTAMP => { + let format = match param(params, "format") { + None => TimestampFormat::default(), + Some(name) => match TimestampFormat::named(&name) { + Some(format) => format, + None => return Some(Err(format!("unknown timestamp format: {name}"))), + }, + }; + text().and_then(|text| { + format + .read(text) + .map(ScalarValue::Timestamp) + .ok_or_else(|| { + format!( + "not a {} timestamp (to the millisecond): {text}", + format.name() + ) + }) + }) + } + _ => return None, + }) +} + +/// The value of parameter `name` among a media type's `params` (`; a=b; c="d"`). +pub(crate) fn param(params: &str, name: &str) -> Option<String> { + params.split(';').find_map(|param| { + let (key, value) = param.split_once('=')?; + key.trim() + .eq_ignore_ascii_case(name) + .then(|| value.trim().trim_matches('"').to_owned()) + }) +} |