summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rw-r--r--src/remake/api.rs91
1 files changed, 85 insertions, 6 deletions
diff --git a/src/remake/api.rs b/src/remake/api.rs
index 51cbe07..8467429 100644
--- a/src/remake/api.rs
+++ b/src/remake/api.rs
@@ -31,8 +31,11 @@
//! `kind::Import`; `GET /admin/document` exports one). Reports leave the document out
//! (`contentBytes`). Requests may be up to 64 MiB.
//!
-//! Restoring a backup, keeping writes made since at their paths:
-//! `base64 -w0 backup.automerge | jq -Rs '{kind: "import", spec: {document: ., rules: [{from: "", to: ""}]}}' | curl --data-binary @- -H 'Content-Type: application/json' …/admin/remakes`
+//! An import can also be uploaded as is, with `Content-Type: application/octet-stream`;
+//! then `rules` (JSON), `baseHeads` (comma-separated), `settleSecs` and
+//! `switchDelaySecs` go in the query. Restoring a backup, keeping writes made since at
+//! their paths:
+//! `curl --data-binary @backup.automerge -H 'Content-Type: application/octet-stream' --url-query 'rules=[{"from":"","to":""}]' …/admin/remakes`
use std::time::Duration;
@@ -41,9 +44,13 @@ use axum::{
Json,
body::Bytes,
extract::{Path, Query, State},
- http::{StatusCode, header::LOCATION},
+ http::{
+ HeaderMap, StatusCode,
+ header::{CONTENT_TYPE, LOCATION},
+ },
response::{IntoResponse, Response},
};
+use base64::engine::{Engine, general_purpose::STANDARD as BASE64};
use serde::{Deserialize, Serialize};
use serde_json::{Value as JsonValue, json};
use tokio::sync::broadcast;
@@ -65,12 +72,64 @@ pub(crate) struct WaitQuery {
}
#[derive(Debug, Default, Deserialize)]
+#[serde(rename_all = "camelCase")]
pub(crate) struct StartQuery {
#[serde(default)]
wait: bool,
/// Allows a settle period shorter than the dormancy threshold.
#[serde(default)]
force: bool,
+ // Only for an import uploaded as `application/octet-stream`, which has no JSON
+ // body to carry them:
+ /// The rules, as JSON (as in the spec).
+ rules: Option<String>,
+ /// Comma-separated hex.
+ base_heads: Option<String>,
+ settle_secs: Option<u64>,
+ switch_delay_secs: Option<u64>,
+}
+
+impl StartQuery {
+ fn has_request_fields(&self) -> bool {
+ self.rules.is_some()
+ || self.base_heads.is_some()
+ || self.settle_secs.is_some()
+ || self.switch_delay_secs.is_some()
+ }
+
+ /// An import request for a document uploaded as is.
+ fn import_request(&self, document: &[u8]) -> Result<StartRequest, String> {
+ let rules = match &self.rules {
+ None => JsonValue::Array(Vec::new()),
+ Some(rules) => serde_json::from_str(rules)
+ .map_err(|error| format!("rules are not valid JSON: {error}"))?,
+ };
+ Ok(StartRequest {
+ kind: Some("import".to_owned()),
+ spec: Some(json!({
+ "document": BASE64.encode(document),
+ "rules": rules,
+ })),
+ base_heads: self.base_heads.as_ref().map(|heads| {
+ heads
+ .split(',')
+ .filter(|head| !head.trim().is_empty())
+ .map(str::to_owned)
+ .collect()
+ }),
+ settle_secs: self.settle_secs,
+ switch_delay_secs: self.switch_delay_secs,
+ ..StartRequest::default()
+ })
+ }
+}
+
+fn is_octet_stream(headers: &HeaderMap) -> bool {
+ headers
+ .get(CONTENT_TYPE)
+ .and_then(|value| value.to_str().ok())
+ .and_then(|value| value.split(';').next())
+ .is_some_and(|media| media.trim() == "application/octet-stream")
}
#[derive(Debug, Default, Deserialize)]
@@ -192,23 +251,26 @@ async fn wait_until_final<H: Host>(host: &H, id: &str) -> Option<Report> {
pub(crate) async fn start<H: Host>(
State(host): State<H>,
Query(query): Query<StartQuery>,
+ headers: HeaderMap,
body: Bytes,
) -> Response {
- start_kind(host, query, body, None).await
+ start_kind(host, query, &headers, body, None).await
}
/// `POST /admin/compaction`: the original endpoint, for compacting remakes only.
pub(crate) async fn start_compaction<H: Host>(
State(host): State<H>,
Query(query): Query<StartQuery>,
+ headers: HeaderMap,
body: Bytes,
) -> Response {
- start_kind(host, query, body, Some(kind::DEFAULT_KIND)).await
+ start_kind(host, query, &headers, body, Some(kind::DEFAULT_KIND)).await
}
async fn start_kind<H: Host>(
host: H,
query: StartQuery,
+ headers: &HeaderMap,
body: Bytes,
only: Option<&'static str>,
) -> Response {
@@ -216,7 +278,24 @@ async fn start_kind<H: Host>(
return error_response(StatusCode::SERVICE_UNAVAILABLE, "node is not ready", None);
}
- let request: StartRequest = if body.is_empty() {
+ let request: StartRequest = if is_octet_stream(headers) {
+ match query.import_request(&body) {
+ Ok(request) => request,
+ Err(reason) => {
+ return error_response(
+ StatusCode::BAD_REQUEST,
+ format!("invalid request: {reason}"),
+ None,
+ );
+ }
+ }
+ } else if query.has_request_fields() {
+ return error_response(
+ StatusCode::BAD_REQUEST,
+ "rules, baseHeads, settleSecs and switchDelaySecs go in the query only for an application/octet-stream import; put them in the JSON body",
+ None,
+ );
+ } else if body.is_empty() {
StartRequest::default()
} else {
match serde_json::from_slice(&body) {