initial commit

4 files changed, 721 insertions, 0 deletions

diff --git a/src/api/debug.rs b/src/api/debug.rs
new file mode 100644
index 0000000..3c24f62
--- /dev/null
+++ b/src/api/debug.rs
@@ -0,0 +1,41 @@
+//! /debug/* handlers (only routed with --debug-api).
+
+use std::collections::HashSet;
+
+use axum::{body::Bytes, extract::State, http::StatusCode, Json};
+use tracing::{info, warn};
+
+use crate::firewall::FirewallRules;
+use crate::state::AppState;
+
+pub(crate) async fn get_debug_firewall(State(state): State<AppState>) -> Json<FirewallRules> {
+    Json(state.firewall.rules())
+}
+
+pub(crate) async fn put_debug_firewall(
+    State(state): State<AppState>,
+    Json(rules): Json<FirewallRules>,
+) -> StatusCode {
+    info!(inbound = ?rules.inbound, outbound = ?rules.outbound, "Firewall rules replaced");
+    state.firewall.set_rules(rules);
+    StatusCode::NO_CONTENT
+}
+
+/// Compatibility for the existing harness: block a set of peers in both directions.
+/// Parses the body without requiring a Content-Type header, as before.
+pub(crate) async fn put_debug_blacklist(State(state): State<AppState>, body: Bytes) -> StatusCode {
+    match serde_json::from_slice::<HashSet<String>>(&body) {
+        Ok(peers) => {
+            info!(peers = ?peers, "Blacklist replaced (both directions)");
+            state.firewall.set_rules(FirewallRules {
+                inbound: peers.clone(),
+                outbound: peers,
+            });
+            StatusCode::NO_CONTENT
+        }
+        Err(error) => {
+            warn!(error = %error, "Invalid blacklist body");
+            StatusCode::BAD_REQUEST
+        }
+    }
+}
diff --git a/src/api/mod.rs b/src/api/mod.rs
new file mode 100644
index 0000000..8038f16
--- /dev/null
+++ b/src/api/mod.rs
@@ -0,0 +1,160 @@
+//! HTTP API: router, OpenAPI document and shared response helpers.
+
+mod debug;
+mod peers;
+mod store;
+
+use axum::{
+    http::{
+        header::{CONTENT_TYPE, ETAG},
+        HeaderValue, StatusCode,
+    },
+    response::{IntoResponse, Response},
+    routing::{delete, get, put},
+    Router,
+};
+use serde_json::Value as JsonValue;
+use sha2::{Digest, Sha256};
+use tracing::{error, warn};
+use utoipa::OpenApi;
+use utoipa_swagger_ui::SwaggerUi;
+
+use crate::auth::{basic_auth, AuthState};
+use crate::state::AppState;
+use crate::store::StoreError;
+
+/// Builds the complete HTTP router. Everything except Swagger is authenticated.
+pub(crate) fn router(state: AppState, auth_state: AuthState, debug_api: bool) -> Router {
+    let mut api: Router<AppState> = Router::new()
+        .route("/store", get(store::get_store_root))
+        .route(
+            "/store/{*key}",
+            get(store::get_store_entry)
+                .put(store::put_store_entry)
+                .delete(store::delete_store_entry),
+        )
+        .route("/stream/{*key}", get(store::stream_store_entry))
+        .route("/peers", get(peers::list_peers))
+        .route("/peers/{peer}", delete(peers::delete_peer));
+
+    if debug_api {
+        warn!("Debug API enabled");
+        api = api
+            .route(
+                "/debug/firewall",
+                get(debug::get_debug_firewall).put(debug::put_debug_firewall),
+            )
+            .route("/debug/blacklist", put(debug::put_debug_blacklist));
+    }
+
+    api.layer(axum::middleware::from_fn_with_state(auth_state, basic_auth))
+        .merge(SwaggerUi::new("/swagger-ui").url("/openapi.json", ApiDoc::openapi()))
+        .with_state(state)
+}
+
+// --- shared helpers ----------------------------------------------------------
+
+pub(crate) fn store_status(error: StoreError, context: &str) -> StatusCode {
+    match error {
+        StoreError::Conflict => StatusCode::CONFLICT,
+        StoreError::NotFound => StatusCode::NOT_FOUND,
+        other => {
+            error!(error = %format!("{other:#}"), "{context}");
+            StatusCode::INTERNAL_SERVER_ERROR
+        }
+    }
+}
+
+fn etag_for(bytes: &[u8]) -> String {
+    let digest = Sha256::digest(bytes);
+    let hex = digest
+        .iter()
+        .map(|byte| format!("{byte:02x}"))
+        .collect::<String>();
+
+    format!("\"{hex}\"")
+}
+
+fn parse_key_path(key: &str) -> Result<Vec<String>, StatusCode> {
+    let segments: Vec<String> = key.split('/').map(str::to_owned).collect();
+
+    if segments.is_empty() || segments.iter().any(|segment| segment.is_empty()) {
+        return Err(StatusCode::BAD_REQUEST);
+    }
+
+    Ok(segments)
+}
+
+fn response_for_value(
+    value: &JsonValue,
+    accept: Option<&HeaderValue>,
+) -> Result<Response, StatusCode> {
+    let wants_text = accept
+        .and_then(|value| value.to_str().ok())
+        .is_some_and(|value| {
+            value
+                .split(',')
+                .any(|item| item.trim().starts_with("text/plain"))
+        });
+
+    let (content_type, body) = if wants_text {
+        let body = match value {
+            JsonValue::String(value) => value.clone(),
+            other => serde_json::to_string(other).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?,
+        };
+
+        ("text/plain; charset=utf-8", body)
+    } else {
+        (
+            "application/json",
+            serde_json::to_string(value).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?,
+        )
+    };
+
+    let etag = etag_for(body.as_bytes());
+
+    let mut response = body.into_response();
+
+    response
+        .headers_mut()
+        .insert(CONTENT_TYPE, HeaderValue::from_static(content_type));
+
+    response.headers_mut().insert(
+        ETAG,
+        HeaderValue::from_str(&etag).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?,
+    );
+
+    Ok(response)
+}
+
+// --- openapi spec ------------------------------------------------------------
+// TODO spec is mostly out of sync
+
+#[derive(OpenApi)]
+#[openapi(
+    paths(
+        store::get_store_root,
+        store::get_store_entry,
+        store::put_store_entry,
+        store::delete_store_entry,
+        peers::list_peers,
+        peers::delete_peer,
+    ),
+    components(
+        schemas(
+            peers::PeerStatus,
+            peers::PeerListResponse,
+            peers::PeerLivenessStatus,
+        )
+    ),
+    tags(
+        (name = "Store", description = "Key-value store operations"),
+        (name = "Peers", description = "Peer discovery and retirement")
+    ),
+    info(
+        title = "Automerge Replication API",
+        version = "1.0.0",
+        description = "HTTP API for a replicated Automerge key-value store"
+    )
+)]
+struct ApiDoc;
diff --git a/src/api/peers.rs b/src/api/peers.rs
new file mode 100644
index 0000000..b2658a5
--- /dev/null
+++ b/src/api/peers.rs
@@ -0,0 +1,161 @@
+//! /peers handlers.
+
+use std::sync::Arc;
+
+use axum::{
+    extract::{Path, Query, State},
+    http::StatusCode,
+    Json,
+};
+use serde::{Deserialize, Serialize};
+use tokio::sync::Mutex;
+use tokio::time::Instant;
+use tracing::warn;
+use utoipa::ToSchema;
+
+use super::store_status;
+use crate::peer::forgotten_path;
+use crate::state::{unix_millis, AppState, HelloState, Liveness, PeerMembership, PeerState};
+use crate::store;
+
+#[derive(Debug, Serialize, ToSchema)]
+#[serde(rename_all = "lowercase")]
+pub(crate) enum PeerLivenessStatus {
+    Active,
+    Dormant,
+}
+
+#[derive(Debug, Serialize, ToSchema)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct PeerStatus {
+    public_key: String,
+    validated_cluster: Option<String>,
+    status: PeerLivenessStatus,
+    handshake_complete: bool,
+    last_seen_secs_ago: Option<u64>,
+    dormant_for_secs: Option<u64>,
+    has_pending_delivery: bool,
+}
+
+#[derive(Debug, Serialize, ToSchema)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct PeerListResponse {
+    peers: Vec<PeerStatus>,
+}
+
+#[derive(Debug, Deserialize)]
+pub(crate) struct DeletePeerQuery {
+    #[serde(default)]
+    force: bool,
+}
+
+#[utoipa::path(
+    get,
+    path = "/peers",
+    tag = "Peers",
+    responses(
+        (
+            status = 200,
+            description = "Lists known peers and their local replication status",
+            body = PeerListResponse
+        )
+    )
+)]
+pub(crate) async fn list_peers(
+    State(state): State<AppState>,
+) -> Result<Json<PeerListResponse>, StatusCode> {
+    let snapshot: Vec<(String, Arc<Mutex<PeerState>>)> = state
+        .peer_states
+        .read()
+        .await
+        .iter()
+        .map(|(pk, m)| (pk.clone(), Arc::clone(m)))
+        .collect();
+
+    let now = Instant::now();
+    let mut peers = Vec::with_capacity(snapshot.len());
+
+    for (public_key, peer_state_mutex) in snapshot {
+        let ps = peer_state_mutex.lock().await;
+
+        let (status, dormant_for_secs) = match ps.liveness {
+            Liveness::Active => (PeerLivenessStatus::Active, None),
+            Liveness::Dormant { since } => (
+                PeerLivenessStatus::Dormant,
+                Some(now.saturating_duration_since(since).as_secs()),
+            ),
+        };
+
+        peers.push(PeerStatus {
+            public_key,
+            validated_cluster: match &ps.membership {
+                PeerMembership::Unknown => None,
+                PeerMembership::Member { cluster_id } => Some(cluster_id.clone()),
+            },
+            status,
+            handshake_complete: ps.hello_state == HelloState::Established,
+            last_seen_secs_ago: ps
+                .last_seen
+                .map(|t| now.saturating_duration_since(t).as_secs()),
+            dormant_for_secs,
+            has_pending_delivery: ps.pending_outbound_sync.is_some(),
+        });
+    }
+
+    peers.sort_by(|a, b| a.public_key.cmp(&b.public_key));
+    Ok(Json(PeerListResponse { peers }))
+}
+
+#[utoipa::path(
+    delete,
+    path = "/peers/{peer}",
+    tag = "Peers",
+    params(
+        ("peer" = String, Path, description = "Peer public key"),
+        ("force" = Option<bool>, Query, description = "Also forget a peer that is currently alive; it returns on its next message")
+    ),
+    responses(
+        (status = 204, description = "Cluster-wide forget recorded"),
+        (status = 400, description = "Not a public key, or the local node's own key"),
+        (status = 409, description = "Peer is alive; pass force=true"),
+        (status = 503, description = "Node is not ready")
+    )
+)]
+pub(crate) async fn delete_peer(
+    State(state): State<AppState>,
+    Path(peer): Path<String>,
+    Query(query): Query<DeletePeerQuery>,
+) -> Result<StatusCode, StatusCode> {
+    if !state.is_ready().await {
+        return Err(StatusCode::SERVICE_UNAVAILABLE);
+    }
+    if peer.is_empty() || peer == state.local_id() {
+        return Err(StatusCode::BAD_REQUEST);
+    }
+
+    // Unknown locally is fine: other nodes may still know the peer.
+    if let Some(peer_state_mutex) = state.get_peer_state(&peer).await {
+        if peer_state_mutex
+            .lock()
+            .await
+            .is_alive(&state.timing, Instant::now())
+            && !query.force
+        {
+            return Err(StatusCode::CONFLICT);
+        }
+    }
+
+    let record = serde_json::json!({ "at": unix_millis(), "by": state.local_id() });
+
+    let peer_clone = peer.clone();
+    state
+        .mutate_doc(move |doc| store::write_path(doc, &forgotten_path(&peer_clone), &record))
+        .await
+        .map_err(|e| store_status(e, "Failed to write forget record"))?;
+
+    if state.peer_states.write().await.remove(&peer).is_some() {
+        warn!(peer = %peer, "Peer forgotten cluster-wide by operator");
+    }
+
+    Ok(StatusCode::NO_CONTENT)
+}
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))
+}