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.rs51
1 files changed, 23 insertions, 28 deletions
diff --git a/src/api/store.rs b/src/api/store.rs
index a74d968..bd4ae8e 100644
--- a/src/api/store.rs
+++ b/src/api/store.rs
@@ -20,7 +20,7 @@ use tokio::sync::broadcast;
use tracing::{debug, error, instrument, warn};
use super::auth::AuthenticatedUser;
-use super::store_status;
+use super::{ApiError, store_status};
use crate::document::meta;
use crate::document::store::{self, StoreError, paths_related};
use crate::state::{AppState, ReadSource};
@@ -48,7 +48,7 @@ pub(crate) struct DeleteQuery {
fn conflicts_response(
value: JsonValue,
conflicts: Vec<store::ConflictReport>,
-) -> Result<Response, StatusCode> {
+) -> Result<Response, ApiError> {
let body = serde_json::json!({ "value": value, "conflicts": conflicts });
response_for_value(&body, None) // always JSON
}
@@ -58,7 +58,7 @@ pub(crate) async fn get_store_root(
State(state): State<AppState>,
headers: HeaderMap,
Query(query): Query<ReadQuery>,
-) -> Result<Response, StatusCode> {
+) -> Result<Response, ApiError> {
read(&state, Vec::new(), &headers, &query).await
}
@@ -69,7 +69,7 @@ pub(crate) async fn get_store_entry(
headers: HeaderMap,
Query(query): Query<ReadQuery>,
RawQuery(raw_query): RawQuery,
-) -> Result<Response, StatusCode> {
+) -> Result<Response, ApiError> {
let path = parse_key_path(&key)?;
if let Some(redirect) = to_positive(&state, &path, false, raw_query.as_deref()).await? {
return Ok(redirect);
@@ -85,7 +85,7 @@ async fn read(
path: Vec<String>,
headers: &HeaderMap,
query: &ReadQuery,
-) -> Result<Response, StatusCode> {
+) -> Result<Response, ApiError> {
let root = path.is_empty();
if query.shallow {
return shallow(state, path, root, query).await;
@@ -110,7 +110,7 @@ async fn read(
let (value, conflicts) = match found {
Some(found) => found,
None if root => (JsonValue::Object(Default::default()), Some(Vec::new())),
- None => return Err(StatusCode::NOT_FOUND),
+ None => return Err(StatusCode::NOT_FOUND.into()),
};
match conflicts {
Some(conflicts) => conflicts_response(value, conflicts),
@@ -129,9 +129,9 @@ async fn shallow(
path: Vec<String>,
unready_live: bool,
query: &ReadQuery,
-) -> Result<Response, StatusCode> {
+) -> Result<Response, ApiError> {
if query.conflicts {
- return Err(StatusCode::BAD_REQUEST);
+ return Err(StatusCode::BAD_REQUEST.into());
}
let (response, source, heads) = state
.render_read(
@@ -157,10 +157,10 @@ const STALE_HEADER: &str = "x-continuum-stale";
const HEADS_HEADER: &str = "x-continuum-heads";
fn mark_source(
- response: Result<Response, StatusCode>,
+ response: Result<Response, ApiError>,
source: ReadSource,
heads: Option<Vec<ChangeHash>>,
-) -> Result<Response, StatusCode> {
+) -> Result<Response, ApiError> {
let mut response = response?;
if let Some(heads) = heads {
let joined = heads
@@ -188,9 +188,9 @@ pub(crate) async fn put_store_entry(
RawQuery(raw_query): RawQuery,
headers: HeaderMap,
body: Bytes,
-) -> Result<Response, StatusCode> {
+) -> Result<Response, ApiError> {
if !state.is_ready().await {
- return Err(StatusCode::SERVICE_UNAVAILABLE);
+ return Err(StatusCode::SERVICE_UNAVAILABLE.into());
}
let path = parse_write_path(&state, &user, &key, "PUT")?;
@@ -257,7 +257,7 @@ fn parse_write_path(
fn response_for_value(
value: &JsonValue,
accept: Option<&HeaderValue>,
-) -> Result<Response, StatusCode> {
+) -> Result<Response, ApiError> {
let wants_text = accept
.and_then(|value| value.to_str().ok())
.is_some_and(|value| {
@@ -328,9 +328,9 @@ pub(crate) async fn post_store_entry(
RawQuery(raw_query): RawQuery,
headers: HeaderMap,
body: Bytes,
-) -> Result<Response, StatusCode> {
+) -> Result<Response, ApiError> {
if !state.is_ready().await {
- return Err(StatusCode::SERVICE_UNAVAILABLE);
+ return Err(StatusCode::SERVICE_UNAVAILABLE.into());
}
let path = parse_write_path(&state, &user, &key, "POST")?;
if let Some(redirect) = to_positive(&state, &path, true, raw_query.as_deref()).await? {
@@ -347,9 +347,6 @@ pub(crate) async fn post_store_entry(
axum::Json(serde_json::json!({ "index": index })),
)
.into_response()),
- Err(error @ (StoreError::NotAList | StoreError::PastTheEnd { .. })) => {
- Ok((StatusCode::CONFLICT, error.to_string()).into_response())
- }
Err(error) => Err(store_status(error, "Failed to insert into a list")),
}
}
@@ -362,17 +359,17 @@ pub(crate) async fn delete_store_entry(
Path(key): Path<String>,
Query(query): Query<DeleteQuery>,
RawQuery(raw_query): RawQuery,
-) -> Result<Response, StatusCode> {
+) -> Result<Response, ApiError> {
let path = parse_write_path(&state, &user, &key, "DELETE")?;
tracing::Span::current().record("path", path.join("/").as_str());
if !state.is_ready().await {
- return Err(StatusCode::SERVICE_UNAVAILABLE);
+ return Err(StatusCode::SERVICE_UNAVAILABLE.into());
}
if path.is_empty() {
- return Err(StatusCode::BAD_REQUEST);
+ return Err(StatusCode::BAD_REQUEST.into());
}
if let Some(redirect) = to_positive(&state, &path, false, raw_query.as_deref()).await? {
@@ -396,7 +393,7 @@ async fn to_positive(
path: &[String],
insertion: bool,
query: Option<&str>,
-) -> Result<Option<Response>, StatusCode> {
+) -> Result<Option<Response>, ApiError> {
if !path.iter().any(|segment| segment.starts_with('-')) {
return Ok(None);
}
@@ -456,9 +453,7 @@ async fn render_traced(
/// `GET /stream`: follows the whole store, as `GET /store` reads it (events have an
/// empty `key`).
-pub(crate) async fn stream_store_root(
- State(state): State<AppState>,
-) -> Result<Response, StatusCode> {
+pub(crate) async fn stream_store_root(State(state): State<AppState>) -> Result<Response, ApiError> {
stream(state, String::new(), Vec::new()).await
}
@@ -466,16 +461,16 @@ pub(crate) async fn stream_store_root(
pub(crate) async fn stream_store_entry(
State(state): State<AppState>,
Path(key): Path<String>,
-) -> Result<Response, StatusCode> {
+) -> Result<Response, ApiError> {
let path = parse_key_path(&key)?;
stream(state, key, path).await
}
/// Newline-delimited JSON events for the value at `path` (`key` as the client wrote it):
/// the current value, then one whenever the resolved value changes.
-async fn stream(state: AppState, key: String, path: Vec<String>) -> Result<Response, StatusCode> {
+async fn stream(state: AppState, key: String, path: Vec<String>) -> Result<Response, ApiError> {
if !state.is_ready().await {
- return Err(StatusCode::SERVICE_UNAVAILABLE);
+ return Err(StatusCode::SERVICE_UNAVAILABLE.into());
}
// Subscribe first, so nothing between the initial read and the loop is missed.