1 files changed, 100 insertions, 15 deletions

diff --git a/src/remake/api.rs b/src/remake/api.rs
index fbd5f54..8467429 100644
--- a/src/remake/api.rs
+++ b/src/remake/api.rs
@@ -24,22 +24,35 @@
 //! `?force=true` accepts `settleSecs` below the dormancy threshold and `switchDelaySecs`
 //! below one keepalive interval: for tests and debugging, not for real swarms.
 //!
-//! `kind` is `compact` (no spec) or `replace`, whose spec is
+//! `kind` is `compact` (no spec), `replace`, whose spec is
 //! `{ "document": {…store root…}, "rules": [{"from": "a", "to": "b"}, {"from": "c", "drop": true}] }`
-//! (see `kind::Replace`). Reports leave the replacement document out (`contentBytes`).
+//! (see `kind::Replace`), or `import`, whose spec is
+//! `{ "document": "<base64 of an Automerge document>", "rules": […] }` (see
+//! `kind::Import`; `GET /admin/document` exports one). Reports leave the document out
+//! (`contentBytes`). Requests may be up to 64 MiB.
+//!
+//! 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;
 
 use automerge::ChangeHash;
 use axum::{
+    Json,
     body::Bytes,
     extract::{Path, Query, State},
-    http::{header::LOCATION, StatusCode},
+    http::{
+        HeaderMap, StatusCode,
+        header::{CONTENT_TYPE, LOCATION},
+    },
     response::{IntoResponse, Response},
-    Json,
 };
+use base64::engine::{Engine, general_purpose::STANDARD as BASE64};
 use serde::{Deserialize, Serialize};
-use serde_json::{json, Value as JsonValue};
+use serde_json::{Value as JsonValue, json};
 use tokio::sync::broadcast;
 use tracing::{error, info, warn};
 use uuid::Uuid;
@@ -47,9 +60,9 @@ use uuid::Uuid;
 use super::coordinator;
 use super::local::LocalStatus;
 use super::records::{
-    self, unix_millis, AbortRequest, EpochRecord, Phase, Policy, Proposal, Record, Retention,
+    self, AbortRequest, EpochRecord, Phase, Policy, Proposal, Record, Retention, unix_millis,
 };
-use super::{kind, Host};
+use super::{Host, kind};
 use crate::document::store::StoreError;
 
 #[derive(Debug, Default, Deserialize)]
@@ -59,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)]
@@ -186,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 {
@@ -210,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) {
@@ -220,7 +305,7 @@ async fn start_kind<H: Host>(
                     StatusCode::BAD_REQUEST,
                     format!("invalid request: {error}"),
                     None,
-                )
+                );
             }
         }
     };
@@ -247,7 +332,7 @@ async fn start_kind<H: Host>(
                 StatusCode::BAD_REQUEST,
                 format!("invalid {kind_name} spec: {reason}"),
                 None,
-            )
+            );
         }
     };
 
@@ -339,7 +424,7 @@ async fn start_kind<H: Host>(
     let id = Uuid::new_v4().to_string();
     let grace = grace(&host);
 
-    let mut proposal = Proposal {
+    let proposal = Proposal {
         coordinator: host.node_id(),
         from_epoch: host.current_epoch().into(),
         base_heads: Vec::new(),
@@ -559,14 +644,14 @@ async fn abort_by_id<H: Host>(host: H, id: String, wait: bool) -> Response {
                 StatusCode::CONFLICT,
                 "already committed; the switch cannot be undone",
                 Some(json!({ "remake": current })),
-            )
+            );
         }
         Phase::Aborted | Phase::Expired => {
             return error_response(
                 StatusCode::CONFLICT,
                 "already over",
                 Some(json!({ "remake": current })),
-            )
+            );
         }
     }