From 9b976a6c4add3719b8f9afef53578ff778b50aa1 Mon Sep 17 00:00:00 2001 From: tv Date: Sun, 1 Nov 2026 22:29:54 +0100 Subject: api: add support for counters --- docs/openapi.yaml | 25 +++++++++++++++++++--- src/api/mod.rs | 7 ++++--- src/api/store.rs | 27 ++++++++++++++++++++++-- src/document/store.rs | 57 +++++++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 108 insertions(+), 8 deletions(-) diff --git a/docs/openapi.yaml b/docs/openapi.yaml index cc898e6..fac802d 100644 --- a/docs/openapi.yaml +++ b/docs/openapi.yaml @@ -139,14 +139,27 @@ paths: description: 'Node is not ready (and isn''t rejoining: a rejoining node takes writes into its document before the rejoin, carried over when it ends).' post: tags: [Store] - summary: Insert into a list + summary: Insert into a list, or increment a counter description: | Inserts the value (as `PUT` takes it) into a list: the path ends in the index it goes to (`queue/2`), the items from there on moving up one; the list's length, or `-1` (a `307` to it), appends. One request is one Automerge change. To add a key to a map, `PUT` it; to overwrite an item, `PUT` its index. + + With `increment`, adds it to the counter at the path instead (no body): one + is made where there is nothing (in a map; a list's items are only + incremented, `-1` naming the last). Increments of a counter on several nodes + add up. Answers `200` with the counter's value then. Reads show a counter as + a number; a `PUT` there makes it a plain number again. + parameters: + - name: increment + in: query + description: How much to add (negative to subtract). + schema: + type: integer + format: int64 requestBody: - required: true + required: false content: application/json: schema: {} @@ -154,6 +167,12 @@ paths: schema: type: string responses: + '200': + description: With `increment`, the counter's value after it. + content: + application/json: + schema: + type: integer '201': description: Inserted, and where. content: @@ -175,7 +194,7 @@ paths: '307': description: A negative index, made positive (`Location`). '409': - description: Not a list there, or the index is past its end (the message says which, and the length). + description: Not a list there, or the index is past its end (the message says which, and the length); with `increment`, not a counter there. '415': description: Content-Type is neither `application/json` nor `text/plain` (absent means JSON). '503': 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 @@ -77,6 +77,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, +} + #[derive(Debug, Deserialize)] pub(crate) struct DeleteQuery { #[serde(default)] @@ -355,11 +362,14 @@ fn body_value(headers: &HeaderMap, body: &Bytes) -> Result, Extension(user): Extension, Path(key): Path, + Query(query): Query, 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

Result { + 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 { + 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 -- cgit v1.3.1