1 files changed, 359 insertions, 0 deletions
diff --git a/src/api/store.rs b/src/api/store.rs
new file mode 100644
index 0000000..dd5be94
--- /dev/null
+++ b/src/api/store.rs
@@ -0,0 +1,359 @@
+//! /store and /stream handlers.
+
+use std::io::Error as IoError;
+
+use axum::{
+    body::{Body, Bytes},
+    extract::{Path, Query, State},
+    http::{
+        header::{ACCEPT, CONTENT_TYPE},
+        HeaderMap, HeaderValue, StatusCode,
+    },
+    response::{IntoResponse, Response},
+};
+use serde::Deserialize;
+use serde_json::Value as JsonValue;
+use tokio::sync::broadcast;
+use tracing::{debug, error, instrument};
+
+use super::{parse_key_path, response_for_value, store_status};
+use crate::state::AppState;
+use crate::store::{self, StoreError};
+use crate::sync::paths_related;
+
+#[derive(Debug, Default, Deserialize)]
+pub(crate) struct ReadQuery {
+    #[serde(default)]
+    conflicts: bool,
+}
+
+#[derive(Debug, Deserialize)]
+pub(crate) struct DeleteQuery {
+    #[serde(default)]
+    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
+}
+
+#[utoipa::path(
+    get,
+    path = "/store",
+    tag = "Store",
+    params(
+        ("conflicts" = Option<bool>, Query, description = "Wrap the value and list every conflict resolved on the way"),
+    ),
+    responses(
+        (status = 200, description = "Returns the complete store as JSON or text, depending on Accept"),
+        (status = 503, description = "Node is not ready"),
+        (status = 500, description = "Failed to read or serialize the document")
+    )
+)]
+pub(crate) async fn get_store_root(
+    State(state): State<AppState>,
+    headers: HeaderMap,
+    Query(query): Query<ReadQuery>,
+) -> Result<Response, StatusCode> {
+    let accept = headers.get(ACCEPT).cloned();
+    let want_conflicts = query.conflicts;
+
+    state
+        .render(
+            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)))
+                }
+            },
+            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()),
+                }
+            },
+        )
+        .await
+}
+
+#[utoipa::path(
+    get,
+    path = "/store/{key}",
+    tag = "Store",
+    params(
+        ("key" = String, Path, description = "Slash-separated store path, for example settings/theme"),
+        ("conflicts" = Option<bool>, Query, description = "Wrap the value and list every conflict resolved on the way"),
+    ),
+    responses(
+        (status = 200, description = "Returns the value as JSON or text, depending on Accept"),
+        (status = 400, description = "Invalid store path"),
+        (status = 404, description = "Value does not exist"),
+        (status = 500, description = "Failed to read or serialize the value"),
+        (status = 503, description = "Node is not ready")
+    )
+)]
+pub(crate) async fn get_store_entry(
+    State(state): State<AppState>,
+    Path(key): Path<String>,
+    headers: HeaderMap,
+    Query(query): Query<ReadQuery>,
+) -> Result<Response, StatusCode> {
+    if !state.is_ready().await {
+        return Err(StatusCode::SERVICE_UNAVAILABLE);
+    }
+
+    let path = parse_key_path(&key)?;
+
+    let accept = headers.get(ACCEPT).cloned();
+    let want_conflicts = query.conflicts;
+
+    state
+        .render(
+            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 |result| {
+                let (value, conflicts) = 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()),
+                }
+            },
+        )
+        .await
+}
+
+#[utoipa::path(
+    put,
+    path = "/store/{key}",
+    tag = "Store",
+    params(
+        ("key" = String, Path, description = "Slash-separated store path")
+    ),
+    request_body(
+        content = String,
+        description = "Value to write. Use application/json for JSON values or text/plain for strings.",
+        content_type = "application/json"
+    ),
+    responses(
+        (status = 200, description = "Value written successfully"),
+        (status = 400, description = "Invalid path, malformed JSON, invalid UTF-8, or unsupported JSON value"),
+        (status = 409, description = "Path conflicts with an existing non-map value"),
+        (status = 415, description = "Unsupported Content-Type"),
+        (status = 500, description = "Internal error or synchronization notification failure"),
+        (status = 503, description = "Node is not ready")
+    )
+)]
+pub(crate) async fn put_store_entry(
+    State(state): State<AppState>,
+    Path(key): Path<String>,
+    headers: HeaderMap,
+    body: Bytes,
+) -> Result<Response, StatusCode> {
+    if !state.is_ready().await {
+        return Err(StatusCode::SERVICE_UNAVAILABLE);
+    }
+
+    let path = parse_key_path(&key)?;
+
+    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)?,
+        ),
+        _ => return Err(StatusCode::UNSUPPORTED_MEDIA_TYPE),
+    };
+
+    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))
+}
+
+#[utoipa::path(
+    delete,
+    path = "/store/{key}",
+    tag = "Store",
+    params(
+        ("key" = String, Path, description = "Slash-separated store path"),
+        ("prune" = Option<bool>, Query, description = "Remove empty ancestor maps after deletion")
+    ),
+    responses(
+        (status = 204, description = "Value deleted successfully"),
+        (status = 400, description = "Invalid store path"),
+        (status = 404, description = "Value does not exist"),
+        (status = 500, description = "Deletion or synchronization notification failed"),
+        (status = 503, description = "Node is not ready")
+    )
+)]
+#[instrument(skip(state), fields(path, prune = query.prune))]
+pub(crate) async fn delete_store_entry(
+    State(state): State<AppState>,
+    Path(key): Path<String>,
+    Query(query): Query<DeleteQuery>,
+) -> Result<StatusCode, StatusCode> {
+    let path = parse_key_path(&key)?;
+
+    tracing::Span::current().record("path", path.join("/").as_str());
+
+    if !state.is_ready().await {
+        return Err(StatusCode::SERVICE_UNAVAILABLE);
+    }
+
+    if path.is_empty() {
+        return Err(StatusCode::BAD_REQUEST);
+    }
+
+    let prune = query.prune;
+    state
+        .mutate_doc(move |doc| store::delete_path(doc, &path, prune))
+        .await
+        .map_err(|e| store_status(e, "Failed to delete store value"))?;
+
+    Ok(StatusCode::NO_CONTENT)
+}
+
+// --- streams -----------------------------------------------------------------
+
+async fn render_traced(
+    state: &AppState,
+    path: Vec<String>,
+) -> Result<(Option<JsonValue>, bool), StoreError> {
+    state
+        .render(move |doc| store::read_path_traced(doc, &path), |result| result)
+        .await
+}
+
+pub(crate) async fn stream_store_entry(
+    State(state): State<AppState>,
+    Path(key): Path<String>,
+) -> Result<Response, StatusCode> {
+    if !state.is_ready().await {
+        return Err(StatusCode::SERVICE_UNAVAILABLE);
+    }
+
+    let path = parse_key_path(&key)?;
+
+    // Subscribe first, so nothing between the initial read and the loop is missed.
+    let mut changes = state.change_events.subscribe();
+
+    let initial = render_traced(&state, path.clone())
+        .await
+        .map_err(|e| store_status(e, "Failed to read initial stream value"))?;
+
+    let stream_state = state.clone();
+    let stream_key = key.clone();
+
+    let output = async_stream::stream! {
+        let (mut last_value, mut conflicted) = initial;
+
+        match stream_event_line(&stream_key, last_value.as_ref()) {
+            Ok(line) => yield Ok::<Bytes, IoError>(line),
+            Err(error) => {
+                yield Err(error);
+                return;
+            }
+        }
+
+        loop {
+            let relevant = match changes.recv().await {
+                // Conflicted paths can change without a patch on their path; re-read always.
+                Ok(set) => conflicted || set.paths.iter().any(|changed| paths_related(changed, &path)),
+                Err(broadcast::error::RecvError::Lagged(skipped)) => {
+                    debug!(skipped, key = %stream_key, "Stream lagged; re-reading");
+                    true
+                }
+                Err(broadcast::error::RecvError::Closed) => break,
+            };
+
+            if !relevant {
+                continue;
+            }
+
+            let (next_value, next_conflicted) = match render_traced(&stream_state, path.clone()).await {
+                Ok(result) => result,
+                Err(error) => {
+                    error!(error = %format!("{error:#}"), key = %stream_key, "Failed to read updated stream value");
+                    yield Err(IoError::other("Failed to read updated store value"));
+                    return;
+                }
+            };
+            conflicted = next_conflicted;
+
+            // A related change can still leave the resolved value unchanged.
+            if next_value == last_value {
+                continue;
+            }
+
+            match stream_event_line(&stream_key, next_value.as_ref()) {
+                Ok(line) => yield Ok(line),
+                Err(error) => {
+                    yield Err(error);
+                    return;
+                }
+            }
+            last_value = next_value;
+        }
+    };
+
+    let mut response = Body::from_stream(output).into_response();
+    response.headers_mut().insert(
+        CONTENT_TYPE,
+        HeaderValue::from_static("application/x-ndjson"),
+    );
+    response.headers_mut().insert(
+        axum::http::header::CACHE_CONTROL,
+        HeaderValue::from_static("no-cache, no-transform"),
+    );
+    Ok(response)
+}
+
+fn stream_event_line(key: &str, value: Option<&JsonValue>) -> std::io::Result<Bytes> {
+    let mut event = serde_json::Map::new();
+
+    event.insert("key".to_owned(), JsonValue::String(key.to_owned()));
+
+    match value {
+        Some(value) => {
+            event.insert("value".to_owned(), value.clone());
+        }
+        None => {
+            event.insert("deleted".to_owned(), JsonValue::Bool(true));
+        }
+    }
+
+    let mut line = serde_json::to_vec(&JsonValue::Object(event)).map_err(IoError::other)?;
+
+    line.push(b'\n');
+    Ok(Bytes::from(line))
+}