diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/api/mod.rs | 7 | ||||
| -rw-r--r-- | src/api/store.rs | 27 | ||||
| -rw-r--r-- | src/document/store.rs | 57 |
3 files changed, 86 insertions, 5 deletions
diff --git a/src/api/mod.rs b/src/api/mod.rs index 8ce855e..dfefcc8 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -201,9 +201,10 @@ impl IntoResponse for ApiError { /// client's; others are logged with `context`, and are a bare `500`. pub(crate) fn store_status(error: StoreError, context: &str) -> ApiError { let status = match error { - StoreError::Conflict | StoreError::NotAList | StoreError::PastTheEnd { .. } => { - StatusCode::CONFLICT - } + StoreError::Conflict + | StoreError::NotAList + | StoreError::PastTheEnd { .. } + | StoreError::NotACounter => StatusCode::CONFLICT, StoreError::NotFound | StoreError::UnknownHeads => StatusCode::NOT_FOUND, StoreError::Rejoining => StatusCode::SERVICE_UNAVAILABLE, other => { diff --git a/src/api/store.rs b/src/api/store.rs index 8b09487..7039008 100644 --- a/src/api/store.rs +++ b/src/api/store.rs @@ -78,6 +78,13 @@ macro_rules! read_at { } #[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>, +} + +#[derive(Debug, Deserialize)] pub(crate) struct DeleteQuery { #[serde(default)] prune: bool, @@ -355,11 +362,14 @@ fn body_value(headers: &HeaderMap, body: &Bytes) -> Result<JsonValue, StatusCode /// `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. +/// 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`. 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, @@ -368,9 +378,22 @@ pub(crate) async fn post_store_entry( return Err(StatusCode::SERVICE_UNAVAILABLE.into()); } let path = parse_write_path(&state, &user, &key, "POST")?; - if let Some(redirect) = to_positive(&state, &path, true, None, raw_query.as_deref()).await? { + // An increment names an item (`-1`: the last); an insert where it goes (`-1`: the end). + let insertion = query.increment.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 = 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 response_for_value(&JsonValue::from(value), headers.get(ACCEPT)); + } let index = path.last().and_then(|leaf| store::list_index(leaf)); let value = body_value(&headers, &body)?; let inserted = state diff --git a/src/document/store.rs b/src/document/store.rs index c16767b..13824b9 100644 --- a/src/document/store.rs +++ b/src/document/store.rs @@ -94,6 +94,9 @@ pub(crate) enum StoreError { /// An index past a list's end (`length` items). #[error("index {index} is past the list's end ({length} items)")] PastTheEnd { index: usize, length: usize }, + /// An increment of something that isn't a counter (`increment_path`). + #[error("not a counter")] + NotACounter, /// A read at heads the document doesn't have (`at::At`). #[error("this node doesn't have these heads (yet)")] UnknownHeads, @@ -744,6 +747,60 @@ fn parent_for_write(doc: &mut impl Transactable, parents: &[String]) -> Result<P Ok(current) } +/// Adds `by` to the counter at `path`, creating one with `by` where there is nothing +/// (in a map: a list's items are only incremented); returns its value then. Increments +/// of one counter on several nodes add up, where writes of numbers would leave one. +/// Anything else there (the newest of concurrent values, as a read resolves them) is +/// `NotACounter`. +pub(crate) fn increment_path( + doc: &mut impl Transactable, + path: &[String], + by: i64, +) -> Result<i64, StoreError> { + let (leaf, parents) = path.split_last().ok_or(StoreError::NotFound)?; + let (parent, prop): (ObjId, Prop) = match parent_for_write(doc, parents)? { + Parent::Maps(maps) => { + let holds = candidates(doc, &maps, leaf)?; + if holds.is_empty() { + doc.put(&maps[0], leaf.as_str(), ScalarValue::counter(by))?; + return Ok(by); + } + // The map that holds the newest value. + let newest = &holds[0].1; + let map = maps + .iter() + .find(|map| { + doc.get_all(*map, leaf.as_str()) + .is_ok_and(|all| all.iter().any(|(_, id)| id == newest)) + }) + .cloned() + .ok_or(StoreError::NotFound)?; + (map, Prop::Map(leaf.clone())) + } + Parent::List(list) => { + let index = list_index(leaf).ok_or(StoreError::NotFound)?; + if index >= doc.length(&list) { + return Err(StoreError::NotFound); + } + (list, Prop::Seq(index)) + } + }; + counter_at(doc, &parent, prop.clone())?; + doc.increment(&parent, prop.clone(), by)?; + counter_at(doc, &parent, prop) +} + +/// The value of the counter `prop` of `parent` holds (the newest of its values). +fn counter_at(doc: &impl ReadDoc, parent: &ObjId, prop: Prop) -> Result<i64, StoreError> { + match doc.get(parent, prop)? { + Some((Value::Scalar(scalar), _)) => match scalar.as_ref() { + ScalarValue::Counter(counter) => Ok(i64::from(counter)), + _ => Err(StoreError::NotACounter), + }, + _ => Err(StoreError::NotACounter), + } +} + /// Inserts `value` into a list: `path` ends in the index it goes to (the items from /// there on move up one; the list's length appends). An index past the end is /// `PastTheEnd`; a parent that isn't a list `NotAList`; anything but an index there, or |