use axum::Json;
use axum::http::header::{CONTENT_RANGE, RETRY_AFTER};
use axum::http::{HeaderName, StatusCode};
use axum::response::{IntoResponse, Response};
use notedthat_core::{Error as CoreError, StorageError};
use serde::Serialize;
#[derive(Debug, thiserror::Error)]
pub enum ApiError {
#[error("unauthorized")]
Unauthorized,
#[error("forbidden")]
Forbidden,
#[error("indexer upsert backpressure")]
IndexerBackpressureUpsert,
#[error("indexer tombstone backpressure")]
IndexerBackpressureTombstone,
#[error("change event not published after the object was {after}")]
EventPublishFailed {
after: notedthat_write::WriteEffect,
},
#[error("events after {requested} are no longer retained; oldest retained is {oldest}")]
EventsGone {
requested: notedthat_core::EventId,
oldest: notedthat_core::EventId,
},
#[error("event backend unavailable: {message}")]
EventsUnavailable {
message: String,
},
#[error(transparent)]
Core(#[from] CoreError),
#[error(transparent)]
Storage(StorageError),
#[error("precondition failed")]
PreconditionFailed,
#[error("range not satisfiable")]
RangeNotSatisfiable {
complete_length: u64,
},
#[error("not modified")]
NotModified,
#[error("malformed range: {0}")]
MalformedRange(String),
#[error("line range not satisfiable")]
LineRangeNotSatisfiable {
line_total: u64,
byte_total: u64,
},
#[error("no match found for old_string")]
ReplaceNoMatch,
#[error("multiple matches found ({count}); use replace_all to replace them all")]
ReplaceAmbiguous {
count: u64,
},
}
impl From<StorageError> for ApiError {
fn from(e: StorageError) -> Self {
match e {
StorageError::NotModified => Self::NotModified,
StorageError::PreconditionFailed => Self::PreconditionFailed,
StorageError::RangeNotSatisfiable { complete_length } => {
Self::RangeNotSatisfiable { complete_length }
}
other => Self::Storage(other),
}
}
}
impl From<notedthat_write::WriteError> for ApiError {
fn from(e: notedthat_write::WriteError) -> Self {
match e {
notedthat_write::WriteError::Storage(e) => Self::Storage(e),
notedthat_write::WriteError::TooLarge { size, limit }
| notedthat_write::WriteError::PatchTooLarge { size, limit } => {
Self::Core(CoreError::PayloadTooLarge { size, limit })
}
notedthat_write::WriteError::Path(e) => Self::Core(e),
notedthat_write::WriteError::IndexerBackpressureUpsert => {
Self::IndexerBackpressureUpsert
}
notedthat_write::WriteError::IndexerBackpressureTombstone => {
Self::IndexerBackpressureTombstone
}
notedthat_write::WriteError::EventPublishFailed { after } => {
Self::EventPublishFailed { after }
}
notedthat_write::WriteError::PatchLineOutOfRange {
total_lines,
total_bytes,
..
} => Self::LineRangeNotSatisfiable {
line_total: total_lines,
byte_total: total_bytes,
},
notedthat_write::WriteError::PatchInvalidRange { message } => {
Self::Core(CoreError::InvalidInput { message })
}
notedthat_write::WriteError::ReplaceNoMatch => Self::ReplaceNoMatch,
notedthat_write::WriteError::ReplaceAmbiguous { count } => {
Self::ReplaceAmbiguous { count }
}
}
}
}
#[derive(Serialize)]
struct ErrorBody<'a> {
error: &'a str,
message: String,
request_id: String,
}
#[derive(Serialize)]
struct ReplaceAmbiguousBody<'a> {
error: &'a str,
message: String,
request_id: String,
match_count: u64,
}
pub struct ApiErrorResponse {
pub error: ApiError,
pub request_id: String,
}
impl ApiErrorResponse {
#[must_use]
pub fn unauthorized(request_id: String) -> Self {
Self {
error: ApiError::Unauthorized,
request_id,
}
}
}
impl ApiError {
fn status_and_code(&self) -> (StatusCode, &'static str) {
match self {
Self::Unauthorized => (StatusCode::UNAUTHORIZED, "unauthorized"),
Self::Forbidden => (StatusCode::FORBIDDEN, "forbidden"),
Self::IndexerBackpressureUpsert
| Self::IndexerBackpressureTombstone
| Self::EventPublishFailed { .. }
| Self::EventsUnavailable { .. } => {
(StatusCode::SERVICE_UNAVAILABLE, "backend_unavailable")
}
Self::EventsGone { .. } => (StatusCode::GONE, "gone"),
Self::Core(CoreError::InvalidInput { .. }) => {
(StatusCode::BAD_REQUEST, "invalid_request")
}
Self::Core(CoreError::NotFound { .. }) => (StatusCode::NOT_FOUND, "not_found"),
Self::Core(CoreError::PayloadTooLarge { .. }) => {
(StatusCode::PAYLOAD_TOO_LARGE, "payload_too_large")
}
Self::Core(CoreError::MalformedRange(_)) | Self::MalformedRange(_) => {
(StatusCode::BAD_REQUEST, "malformed_range")
}
Self::LineRangeNotSatisfiable { .. }
| Self::Core(CoreError::RangeNotSatisfiable { .. })
| Self::RangeNotSatisfiable { .. } => {
(StatusCode::RANGE_NOT_SATISFIABLE, "range_not_satisfiable")
}
Self::Core(CoreError::NotModified) | Self::NotModified => {
(StatusCode::NOT_MODIFIED, "not_modified")
}
Self::Core(CoreError::PreconditionFailed) | Self::PreconditionFailed => {
(StatusCode::PRECONDITION_FAILED, "precondition_failed")
}
Self::ReplaceNoMatch => (StatusCode::UNPROCESSABLE_ENTITY, "no_match"),
Self::ReplaceAmbiguous { .. } => (StatusCode::UNPROCESSABLE_ENTITY, "ambiguous_match"),
Self::Core(CoreError::BucketNameTooLong { .. } | CoreError::Config { .. }) => {
(StatusCode::INTERNAL_SERVER_ERROR, "internal_error")
}
Self::Core(CoreError::Storage(e)) | Self::Storage(e) => Self::storage_status(e),
}
}
fn storage_status(e: &StorageError) -> (StatusCode, &'static str) {
match e {
StorageError::NotFound { .. } | StorageError::BucketNotFound { .. } => {
(StatusCode::NOT_FOUND, "not_found")
}
StorageError::BackendUnavailable { .. } => {
(StatusCode::SERVICE_UNAVAILABLE, "backend_unavailable")
}
StorageError::NotModified => (StatusCode::NOT_MODIFIED, "not_modified"),
StorageError::PreconditionFailed => {
(StatusCode::PRECONDITION_FAILED, "precondition_failed")
}
StorageError::RangeNotSatisfiable { .. } => {
(StatusCode::RANGE_NOT_SATISFIABLE, "range_not_satisfiable")
}
StorageError::Other { .. } => (StatusCode::INTERNAL_SERVER_ERROR, "internal_error"),
}
}
fn range_not_satisfiable_length(&self) -> Option<u64> {
match self {
Self::RangeNotSatisfiable { complete_length }
| Self::Storage(StorageError::RangeNotSatisfiable { complete_length })
| Self::Core(CoreError::RangeNotSatisfiable { complete_length }) => {
Some(*complete_length)
}
_ => None,
}
}
fn is_not_modified(&self) -> bool {
matches!(
self,
Self::NotModified
| Self::Storage(StorageError::NotModified)
| Self::Core(CoreError::NotModified)
)
}
}
impl ApiErrorResponse {
fn retry_later(request_id: String, message: String) -> Response {
let body = ErrorBody {
error: "backend_unavailable",
message,
request_id,
};
(
StatusCode::SERVICE_UNAVAILABLE,
[(RETRY_AFTER, "5")],
Json(body),
)
.into_response()
}
}
impl IntoResponse for ApiErrorResponse {
fn into_response(self) -> Response {
if let ApiError::LineRangeNotSatisfiable {
line_total,
byte_total,
} = &self.error
{
return (
StatusCode::RANGE_NOT_SATISFIABLE,
[
(CONTENT_RANGE, format!("lines */{line_total}")),
(
HeaderName::from_static("x-content-range-bytes"),
format!("*/{byte_total}"),
),
],
)
.into_response();
}
if let Some(complete_length) = self.error.range_not_satisfiable_length() {
let content_range = format!("bytes */{complete_length}");
return (
StatusCode::RANGE_NOT_SATISFIABLE,
[(CONTENT_RANGE, content_range)],
)
.into_response();
}
if self.error.is_not_modified() {
return StatusCode::NOT_MODIFIED.into_response();
}
match &self.error {
ApiError::Unauthorized => {
let body = ErrorBody {
error: "unauthorized",
message: "provide a valid Bearer token in the Authorization header".to_string(),
request_id: self.request_id,
};
return (StatusCode::UNAUTHORIZED, Json(body)).into_response();
}
ApiError::IndexerBackpressureUpsert => {
return Self::retry_later(
self.request_id,
"object stored; indexer queue full — retry to re-enqueue".to_string(),
);
}
ApiError::IndexerBackpressureTombstone => {
return Self::retry_later(
self.request_id,
"deleted from storage; retry to clear from search index".to_string(),
);
}
ApiError::EventPublishFailed { after } => {
let message = match after {
notedthat_write::WriteEffect::Stored => {
"object stored; change event not published — retry to publish"
}
notedthat_write::WriteEffect::Deleted => {
"deleted from storage; change event not published — retry to publish"
}
};
return Self::retry_later(self.request_id, message.to_string());
}
ApiError::EventsUnavailable { message } => {
return Self::retry_later(
self.request_id,
format!("event backend unavailable: {message}"),
);
}
ApiError::ReplaceAmbiguous { count } => {
let body = ReplaceAmbiguousBody {
error: "ambiguous_match",
message: self.error.to_string(),
request_id: self.request_id,
match_count: *count,
};
return (StatusCode::UNPROCESSABLE_ENTITY, Json(body)).into_response();
}
_ => {}
}
let (status, code) = self.error.status_and_code();
let message = self.error.to_string();
let body = ErrorBody {
error: code,
message,
request_id: self.request_id,
};
(status, Json(body)).into_response()
}
}
impl IntoResponse for ApiError {
fn into_response(self) -> Response {
ApiErrorResponse {
error: self,
request_id: "unknown".to_string(),
}
.into_response()
}
}
#[cfg(test)]
mod tests {
use super::*;
use axum::body::{Body, to_bytes};
use axum::http::Request;
use bytes::Bytes;
use notedthat_core::{ConditionalHeaders, KbSlug, ObjectPath, Storage};
use notedthat_indexer::IndexEvent;
use notedthat_write::WriteError;
use std::collections::BTreeMap;
use std::sync::Arc;
use tower::util::ServiceExt;
const KB: &str = "notes";
const TOKEN: &str = "test-token-abc";
fn router() -> axum::Router {
router_with_max_patchable_size(16 * 1024 * 1024)
}
fn router_with_max_patchable_size(max_patchable_size: u64) -> axum::Router {
let kb = KbSlug::try_new(KB).unwrap();
let mut kbs = BTreeMap::new();
kbs.insert(KB.to_string(), kb);
let (indexer_tx, mut rx) = tokio::sync::mpsc::channel(16);
tokio::spawn(async move { while rx.recv().await.is_some() {} });
crate::router::build_router(crate::state::AppState {
storage: Arc::new(crate::testing::InMemoryStorage::default()),
access_policies: Arc::new(notedthat_core::signed_in_policies(&kbs)),
declared_kbs: Arc::new(kbs),
authenticator: Arc::new(notedthat_core::Authenticator::new(TOKEN)),
max_body_size: 16 * 1024 * 1024,
max_patchable_size,
indexer_tx,
searcher: Arc::new(crate::testing::NoopSearcher),
events: None,
})
}
fn router_with_storage_and_indexer(
storage: Arc<dyn Storage>,
kb: KbSlug,
max_patchable_size: u64,
indexer_tx: tokio::sync::mpsc::Sender<IndexEvent>,
) -> axum::Router {
let mut kbs = BTreeMap::new();
kbs.insert(KB.to_string(), kb);
crate::router::build_router(crate::state::AppState {
storage,
access_policies: Arc::new(notedthat_core::signed_in_policies(&kbs)),
declared_kbs: Arc::new(kbs),
authenticator: Arc::new(notedthat_core::Authenticator::new(TOKEN)),
max_body_size: 16 * 1024 * 1024,
max_patchable_size,
indexer_tx,
searcher: Arc::new(crate::testing::NoopSearcher),
events: None,
})
}
async fn put_object(router: axum::Router, path: &str, body: &'static [u8]) -> String {
let response = router
.oneshot(
Request::builder()
.method("PUT")
.uri(format!("/api/v1/knowledgebases/{KB}/{path}"))
.header("authorization", format!("Bearer {TOKEN}"))
.header(axum::http::header::CONTENT_TYPE, "text/markdown")
.body(Body::from(Bytes::from_static(body)))
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::CREATED);
response
.headers()
.get(axum::http::header::ETAG)
.unwrap()
.to_str()
.unwrap()
.to_string()
}
async fn get_object(router: axum::Router, path: &str) -> Bytes {
let response = router
.oneshot(
Request::builder()
.method("GET")
.uri(format!("/api/v1/knowledgebases/{KB}/{path}"))
.header("authorization", format!("Bearer {TOKEN}"))
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
to_bytes(response.into_body(), usize::MAX).await.unwrap()
}
async fn post_replace(
router: axum::Router,
path: &str,
if_match: &str,
body: &'static [u8],
) -> Response {
router
.oneshot(
Request::builder()
.method("POST")
.uri(format!("/api/v1/knowledgebases/{KB}/replace/{path}"))
.header("authorization", format!("Bearer {TOKEN}"))
.header(axum::http::header::CONTENT_TYPE, "application/json")
.header(axum::http::header::IF_MATCH, if_match)
.body(Body::from(Bytes::from_static(body)))
.unwrap(),
)
.await
.unwrap()
}
async fn object_with_etag(
storage: &crate::testing::InMemoryStorage,
kb: &KbSlug,
path: &str,
body: &'static [u8],
) -> String {
storage
.put_object(
kb,
&ObjectPath::try_from_str(path).unwrap(),
Bytes::from_static(body),
Some("text/markdown"),
ConditionalHeaders::default(),
)
.await
.unwrap()
.etag
.unwrap()
}
async fn assert_invalid_request_response(response: Response) {
assert_eq!(response.status(), StatusCode::BAD_REQUEST);
let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
assert_eq!(json["error"], "invalid_request");
}
#[tokio::test]
async fn test_unauthorized_status_and_body() {
let resp = ApiErrorResponse::unauthorized("req-123".to_string()).into_response();
assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
let body = to_bytes(resp.into_body(), usize::MAX).await.unwrap();
let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
assert_eq!(json["error"], "unauthorized");
assert_eq!(json["request_id"], "req-123");
}
#[tokio::test]
async fn test_not_found_status() {
let err = ApiError::Core(CoreError::NotFound {
resource: "foo".into(),
});
let resp = ApiErrorResponse {
error: err,
request_id: "rid".into(),
}
.into_response();
assert_eq!(resp.status(), StatusCode::NOT_FOUND);
}
#[tokio::test]
async fn test_payload_too_large_status() {
let err = ApiError::Core(CoreError::PayloadTooLarge {
size: 20_000_000,
limit: 16_777_216,
});
let resp = ApiErrorResponse {
error: err,
request_id: "rid".into(),
}
.into_response();
assert_eq!(resp.status(), StatusCode::PAYLOAD_TOO_LARGE);
}
#[tokio::test]
async fn test_request_id_in_body() {
let err = ApiError::Core(CoreError::InvalidInput {
message: "bad".into(),
});
let resp = ApiErrorResponse {
error: err,
request_id: "my-req-id".into(),
}
.into_response();
let body = to_bytes(resp.into_body(), usize::MAX).await.unwrap();
let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
assert_eq!(json["request_id"], "my-req-id");
}
#[tokio::test]
async fn test_indexer_backpressure_upsert_503_body_and_retry_after() {
let resp = ApiErrorResponse {
error: ApiError::IndexerBackpressureUpsert,
request_id: "rid".to_string(),
}
.into_response();
assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE);
assert_eq!(resp.headers().get("retry-after").unwrap(), "5");
let body = to_bytes(resp.into_body(), usize::MAX).await.unwrap();
let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
assert_eq!(json["error"], "backend_unavailable");
assert_eq!(
json["message"],
"object stored; indexer queue full — retry to re-enqueue"
);
assert_eq!(json["request_id"], "rid");
}
#[tokio::test]
async fn test_indexer_backpressure_tombstone_503_body_and_retry_after() {
let resp = ApiErrorResponse {
error: ApiError::IndexerBackpressureTombstone,
request_id: "rid".to_string(),
}
.into_response();
assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE);
assert_eq!(resp.headers().get("retry-after").unwrap(), "5");
let body = to_bytes(resp.into_body(), usize::MAX).await.unwrap();
let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
assert_eq!(json["error"], "backend_unavailable");
assert_eq!(
json["message"],
"deleted from storage; retry to clear from search index"
);
assert_eq!(json["request_id"], "rid");
}
#[test]
fn test_from_write_error_indexer_backpressure() {
assert!(matches!(
ApiError::from(WriteError::IndexerBackpressureUpsert),
ApiError::IndexerBackpressureUpsert
));
assert!(matches!(
ApiError::from(WriteError::IndexerBackpressureTombstone),
ApiError::IndexerBackpressureTombstone
));
}
#[test]
fn test_from_write_error_patch_too_large() {
let api_err = ApiError::from(WriteError::PatchTooLarge {
size: 200 * 1024 * 1024,
limit: 100 * 1024 * 1024,
});
let (status, code) = api_err.status_and_code();
assert_eq!(status.as_u16(), 413);
assert_eq!(code, "payload_too_large");
}
#[test]
fn test_from_write_error_patch_line_out_of_range() {
let api_err = ApiError::from(WriteError::PatchLineOutOfRange {
first: 999,
last: 1000,
total_lines: 20,
total_bytes: 100,
});
let (status, code) = api_err.status_and_code();
assert_eq!(status.as_u16(), 416);
assert_eq!(code, "range_not_satisfiable");
}
#[test]
fn test_from_write_error_patch_invalid_range() {
let api_err = ApiError::from(WriteError::PatchInvalidRange {
message: "test".into(),
});
let (status, code) = api_err.status_and_code();
assert_eq!(status.as_u16(), 400);
assert_eq!(code, "invalid_request");
}
#[test]
fn test_from_write_error_replace_no_match() {
let api_err = ApiError::from(WriteError::ReplaceNoMatch);
let (status, code) = api_err.status_and_code();
assert_eq!(status.as_u16(), 422);
assert_eq!(code, "no_match");
}
#[test]
fn test_from_write_error_replace_ambiguous() {
let api_err = ApiError::from(WriteError::ReplaceAmbiguous { count: 3 });
let (status, code) = api_err.status_and_code();
assert_eq!(status.as_u16(), 422);
assert_eq!(code, "ambiguous_match");
}
#[tokio::test]
async fn test_ambiguous_match_body_includes_match_count() {
let resp = ApiErrorResponse {
error: ApiError::ReplaceAmbiguous { count: 3 },
request_id: "req-1".into(),
}
.into_response();
assert_eq!(resp.status(), StatusCode::UNPROCESSABLE_ENTITY);
let body = to_bytes(resp.into_body(), usize::MAX).await.unwrap();
let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
assert_eq!(json["error"], "ambiguous_match");
assert_eq!(json["match_count"], 3);
assert_eq!(json["request_id"], "req-1");
}
#[tokio::test]
async fn test_no_match_body_omits_match_count() {
let resp = ApiErrorResponse {
error: ApiError::ReplaceNoMatch,
request_id: "req-1".into(),
}
.into_response();
assert_eq!(resp.status(), StatusCode::UNPROCESSABLE_ENTITY);
let body = to_bytes(resp.into_body(), usize::MAX).await.unwrap();
let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
assert_eq!(json["error"], "no_match");
assert!(json.get("match_count").is_none());
}
#[tokio::test]
async fn replace_missing_if_match_returns_400_invalid_request() {
let response = router()
.oneshot(
Request::builder()
.method("POST")
.uri(format!("/api/v1/knowledgebases/{KB}/replace/hello.md"))
.header("authorization", format!("Bearer {TOKEN}"))
.header(axum::http::header::CONTENT_TYPE, "application/json")
.body(Body::from(Bytes::from_static(
br#"{"old_string":"x","new_string":"y"}"#,
)))
.unwrap(),
)
.await
.unwrap();
assert_invalid_request_response(response).await;
}
#[tokio::test]
async fn replace_if_match_star_returns_400() {
let response = router()
.oneshot(
Request::builder()
.method("POST")
.uri(format!("/api/v1/knowledgebases/{KB}/replace/hello.md"))
.header("authorization", format!("Bearer {TOKEN}"))
.header(axum::http::header::CONTENT_TYPE, "application/json")
.header(axum::http::header::IF_MATCH, "*")
.body(Body::from(Bytes::from_static(
br#"{"old_string":"x","new_string":"y"}"#,
)))
.unwrap(),
)
.await
.unwrap();
assert_invalid_request_response(response).await;
}
#[tokio::test]
async fn replace_multi_value_if_match_returns_400() {
let response = router()
.oneshot(
Request::builder()
.method("POST")
.uri(format!("/api/v1/knowledgebases/{KB}/replace/hello.md"))
.header("authorization", format!("Bearer {TOKEN}"))
.header(axum::http::header::CONTENT_TYPE, "application/json")
.header(axum::http::header::IF_MATCH, "\"a\",\"b\"")
.body(Body::from(Bytes::from_static(
br#"{"old_string":"x","new_string":"y"}"#,
)))
.unwrap(),
)
.await
.unwrap();
assert_invalid_request_response(response).await;
}
#[tokio::test]
async fn replace_malformed_json_body_returns_400() {
let response = router()
.oneshot(
Request::builder()
.method("POST")
.uri(format!("/api/v1/knowledgebases/{KB}/replace/hello.md"))
.header("authorization", format!("Bearer {TOKEN}"))
.header(axum::http::header::CONTENT_TYPE, "application/json")
.header(axum::http::header::IF_MATCH, "\"valid-etag\"")
.body(Body::from(Bytes::from_static(b"{invalid json")))
.unwrap(),
)
.await
.unwrap();
assert_invalid_request_response(response).await;
}
#[tokio::test]
async fn replace_missing_old_string_field_returns_400() {
let response = router()
.oneshot(
Request::builder()
.method("POST")
.uri(format!("/api/v1/knowledgebases/{KB}/replace/hello.md"))
.header("authorization", format!("Bearer {TOKEN}"))
.header(axum::http::header::CONTENT_TYPE, "application/json")
.header(axum::http::header::IF_MATCH, "\"valid-etag\"")
.body(Body::from(Bytes::from_static(br#"{"new_string":"y"}"#)))
.unwrap(),
)
.await
.unwrap();
assert_invalid_request_response(response).await;
}
#[tokio::test]
async fn replace_empty_old_string_returns_400() {
let response = router()
.oneshot(
Request::builder()
.method("POST")
.uri(format!("/api/v1/knowledgebases/{KB}/replace/hello.md"))
.header("authorization", format!("Bearer {TOKEN}"))
.header(axum::http::header::CONTENT_TYPE, "application/json")
.header(axum::http::header::IF_MATCH, "\"valid-etag\"")
.body(Body::from(Bytes::from_static(
br#"{"old_string":"","new_string":"y"}"#,
)))
.unwrap(),
)
.await
.unwrap();
assert_invalid_request_response(response).await;
}
#[tokio::test]
async fn replace_single_match_happy_returns_200_with_etag_and_match_count() {
let router = router();
let etag = put_object(router.clone(), "hello.md", b"hello world").await;
let response = post_replace(
router.clone(),
"hello.md",
&etag,
br#"{"old_string":"world","new_string":"planet"}"#,
)
.await;
assert_eq!(response.status(), StatusCode::OK);
assert!(response.headers().get(axum::http::header::ETAG).is_some());
assert_eq!(
response.headers().get("content-location").unwrap(),
"/api/v1/knowledgebases/notes/hello.md"
);
let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
assert!(json["etag"].as_str().is_some());
assert_eq!(json["match_count"], 1);
assert_eq!(json["total_bytes"], 12);
assert_eq!(&get_object(router, "hello.md").await[..], b"hello planet");
}
#[tokio::test]
async fn replace_no_match_returns_422_no_match() {
let router = router();
let etag = put_object(router.clone(), "hello.md", b"hello world").await;
let response = post_replace(
router.clone(),
"hello.md",
&etag,
br#"{"old_string":"nonexistent","new_string":"x"}"#,
)
.await;
assert_eq!(response.status(), StatusCode::UNPROCESSABLE_ENTITY);
let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
assert_eq!(json["error"], "no_match");
assert_eq!(&get_object(router, "hello.md").await[..], b"hello world");
}
#[tokio::test]
async fn replace_ambiguous_returns_422_with_match_count() {
let router = router();
let etag = put_object(router.clone(), "hello.md", b"a b a").await;
let response = post_replace(
router.clone(),
"hello.md",
&etag,
br#"{"old_string":"a","new_string":"Z"}"#,
)
.await;
assert_eq!(response.status(), StatusCode::UNPROCESSABLE_ENTITY);
let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
assert_eq!(json["error"], "ambiguous_match");
assert_eq!(json["match_count"], 2);
assert_eq!(&get_object(router, "hello.md").await[..], b"a b a");
}
#[tokio::test]
async fn replace_all_true_multiple_matches_returns_200_with_count_2() {
let router = router();
let etag = put_object(router.clone(), "hello.md", b"a b a").await;
let response = post_replace(
router.clone(),
"hello.md",
&etag,
br#"{"old_string":"a","new_string":"Z","replace_all":true}"#,
)
.await;
assert_eq!(response.status(), StatusCode::OK);
let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
assert_eq!(json["match_count"], 2);
assert_eq!(json["total_bytes"], 5);
assert_eq!(&get_object(router, "hello.md").await[..], b"Z b Z");
}
#[tokio::test]
async fn replace_stale_etag_returns_412() {
let router = router();
put_object(router.clone(), "hello.md", b"hello world").await;
let response = post_replace(
router,
"hello.md",
"\"stale\"",
br#"{"old_string":"world","new_string":"planet"}"#,
)
.await;
assert_eq!(response.status(), StatusCode::PRECONDITION_FAILED);
}
#[tokio::test]
async fn replace_post_splice_size_over_cap_returns_413() {
let router = router_with_max_patchable_size(20);
let etag = put_object(router.clone(), "hello.md", b"1234567890").await;
let response = post_replace(
router,
"hello.md",
&etag,
br#"{"old_string":"0","new_string":"abcdefghijklmnopqrstuvwxyz"}"#,
)
.await;
assert_eq!(response.status(), StatusCode::PAYLOAD_TOO_LARGE);
let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
assert_eq!(json["error"], "payload_too_large");
}
#[tokio::test]
async fn replace_indexer_backpressure_returns_503_with_retry_after() {
let kb = KbSlug::try_new(KB).unwrap();
let storage = crate::testing::InMemoryStorage::default();
let etag = object_with_etag(&storage, &kb, "hello.md", b"hello world").await;
let (indexer_tx, _rx) = tokio::sync::mpsc::channel(1);
indexer_tx
.try_send(IndexEvent::Upsert {
kb: kb.clone(),
object_key: ObjectPath::try_from_str("queued.md").unwrap(),
etag: "queued".to_string(),
mtime: 0,
})
.unwrap();
let router =
router_with_storage_and_indexer(Arc::new(storage), kb, 16 * 1024 * 1024, indexer_tx);
let response = post_replace(
router,
"hello.md",
&etag,
br#"{"old_string":"world","new_string":"planet"}"#,
)
.await;
assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE);
assert_eq!(response.headers().get("retry-after").unwrap(), "5");
let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
assert_eq!(json["error"], "backend_unavailable");
}
#[tokio::test]
async fn test_precondition_failed_body_shape_unchanged() {
let resp = ApiErrorResponse {
error: ApiError::PreconditionFailed,
request_id: "req-1".into(),
}
.into_response();
assert_eq!(resp.status(), StatusCode::PRECONDITION_FAILED);
let body = to_bytes(resp.into_body(), usize::MAX).await.unwrap();
let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
let object = json.as_object().unwrap();
assert_eq!(object.len(), 3);
assert!(object.contains_key("error"));
assert!(object.contains_key("message"));
assert!(object.contains_key("request_id"));
}
#[tokio::test]
async fn test_precondition_failed_412() {
let resp = ApiError::PreconditionFailed.into_response();
assert_eq!(resp.status(), StatusCode::PRECONDITION_FAILED);
}
#[tokio::test]
async fn test_not_modified_304_empty_body() {
let resp = ApiError::NotModified.into_response();
assert_eq!(resp.status(), StatusCode::NOT_MODIFIED);
let body = to_bytes(resp.into_body(), usize::MAX).await.unwrap();
assert!(body.is_empty(), "304 must have an empty body");
}
#[tokio::test]
async fn test_malformed_range_400() {
let resp = ApiError::MalformedRange("bytes=abc".to_string()).into_response();
assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
}
#[tokio::test]
async fn test_range_not_satisfiable_416_content_range_header() {
let resp = ApiError::RangeNotSatisfiable {
complete_length: 100,
}
.into_response();
assert_eq!(resp.status(), StatusCode::RANGE_NOT_SATISFIABLE);
let cr = resp
.headers()
.get("content-range")
.expect("content-range header must be present on 416");
assert_eq!(cr.to_str().unwrap(), "bytes */100");
let body = to_bytes(resp.into_body(), usize::MAX).await.unwrap();
assert!(body.is_empty(), "416 body must be empty per RFC 7233 §4.4");
}
#[tokio::test]
async fn test_storage_range_not_satisfiable_416_content_range_header() {
let resp = ApiError::Storage(StorageError::RangeNotSatisfiable {
complete_length: 42,
})
.into_response();
assert_eq!(resp.status(), StatusCode::RANGE_NOT_SATISFIABLE);
let cr = resp
.headers()
.get("content-range")
.expect("content-range header must be present on 416");
assert_eq!(cr.to_str().unwrap(), "bytes */42");
}
#[tokio::test]
async fn test_storage_not_modified_304_empty_body() {
let resp = ApiError::Storage(StorageError::NotModified).into_response();
assert_eq!(resp.status(), StatusCode::NOT_MODIFIED);
let body = to_bytes(resp.into_body(), usize::MAX).await.unwrap();
assert!(body.is_empty(), "304 must have an empty body");
}
#[test]
fn test_from_storage_error_not_modified() {
let api_err = ApiError::from(StorageError::NotModified);
assert!(matches!(api_err, ApiError::NotModified));
}
#[test]
fn test_from_storage_error_precondition_failed() {
let api_err = ApiError::from(StorageError::PreconditionFailed);
assert!(matches!(api_err, ApiError::PreconditionFailed));
}
#[test]
fn test_from_storage_error_range_not_satisfiable() {
let api_err = ApiError::from(StorageError::RangeNotSatisfiable {
complete_length: 999,
});
assert!(
matches!(
api_err,
ApiError::RangeNotSatisfiable {
complete_length: 999
}
),
"expected RangeNotSatisfiable with complete_length=999, got {api_err:?}"
);
}
#[test]
fn test_from_storage_error_other_wrapped() {
let api_err = ApiError::from(StorageError::NotFound {
key: "foo".to_string(),
});
assert!(matches!(api_err, ApiError::Storage(_)));
}
mod line_range_error {
use super::*;
use axum::body::Body;
use axum::http::Request;
use bytes::Bytes;
use notedthat_core::KbSlug;
use std::collections::BTreeMap;
use std::sync::Arc;
use tower::util::ServiceExt;
const KB: &str = "notes";
const TOKEN: &str = "test-token-abc";
fn twenty_line_markdown() -> String {
let mut body = String::new();
for line in 1..=20 {
std::fmt::Write::write_fmt(&mut body, format_args!("line {line:02}\n")).unwrap();
}
body
}
fn router() -> axum::Router {
let kb = KbSlug::try_new(KB).unwrap();
let mut kbs = BTreeMap::new();
kbs.insert(KB.to_string(), kb);
let (indexer_tx, mut rx) = tokio::sync::mpsc::channel(16);
tokio::spawn(async move { while rx.recv().await.is_some() {} });
crate::router::build_router(crate::state::AppState {
storage: Arc::new(crate::testing::InMemoryStorage::default()),
access_policies: Arc::new(notedthat_core::signed_in_policies(&kbs)),
declared_kbs: Arc::new(kbs),
authenticator: Arc::new(notedthat_core::Authenticator::new(TOKEN)),
max_body_size: 16 * 1024 * 1024,
max_patchable_size: 16 * 1024 * 1024,
indexer_tx,
searcher: Arc::new(crate::testing::NoopSearcher),
events: None,
})
}
async fn put_ranges_md(router: axum::Router) {
let response = router
.oneshot(
Request::builder()
.method("PUT")
.uri(format!("/api/v1/knowledgebases/{KB}/ranges.md"))
.header("authorization", format!("Bearer {TOKEN}"))
.header(axum::http::header::CONTENT_TYPE, "text/markdown")
.body(Body::from(Bytes::from(twenty_line_markdown())))
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::CREATED);
}
#[tokio::test]
async fn malformed_line_range_returns_json_400() {
let response = ApiError::MalformedRange("lines=abc".into()).into_response();
assert_eq!(response.status(), StatusCode::BAD_REQUEST);
let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
assert_eq!(json["error"], "malformed_range");
}
#[tokio::test]
async fn line_range_not_satisfiable_returns_dual_headers_and_empty_body() {
let response = ApiError::LineRangeNotSatisfiable {
line_total: 20,
byte_total: 100,
}
.into_response();
assert_eq!(response.status(), StatusCode::RANGE_NOT_SATISFIABLE);
assert_eq!(
response.headers().get("content-range").unwrap(),
"lines */20"
);
assert_eq!(
response.headers().get("x-content-range-bytes").unwrap(),
"*/100"
);
let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
assert!(body.is_empty());
}
#[tokio::test]
async fn out_of_range_line_get_returns_dual_headers_and_empty_body() {
let router = router();
put_ranges_md(router.clone()).await;
let response = router
.oneshot(
Request::builder()
.method("GET")
.uri(format!("/api/v1/knowledgebases/{KB}/ranges.md"))
.header("authorization", format!("Bearer {TOKEN}"))
.header(axum::http::header::RANGE, "lines=100-200")
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::RANGE_NOT_SATISFIABLE);
assert_eq!(
response.headers().get("content-range").unwrap(),
"lines */20"
);
assert_eq!(
response.headers().get("x-content-range-bytes").unwrap(),
"*/160"
);
let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
assert!(body.is_empty());
}
#[tokio::test]
async fn byte_range_not_satisfiable_omits_line_byte_header() {
let response = ApiError::RangeNotSatisfiable {
complete_length: 100,
}
.into_response();
assert_eq!(response.status(), StatusCode::RANGE_NOT_SATISFIABLE);
assert_eq!(
response.headers().get("content-range").unwrap(),
"bytes */100"
);
assert!(response.headers().get("x-content-range-bytes").is_none());
}
#[test]
fn patch_line_out_of_range_maps_to_line_range_not_satisfiable() {
let error = ApiError::from(WriteError::PatchLineOutOfRange {
first: 100,
last: 200,
total_lines: 20,
total_bytes: 100,
});
assert!(matches!(
error,
ApiError::LineRangeNotSatisfiable {
line_total: 20,
byte_total: 100
}
));
}
}
}