reorganize all the modules

5 files changed, 150 insertions, 8 deletions

diff --git a/src/api/auth.rs b/src/api/auth.rs
new file mode 100644
index 0000000..3926c70
--- /dev/null
+++ b/src/api/auth.rs
@@ -0,0 +1,141 @@
+use argon2::{password_hash::PasswordVerifier, Argon2};
+use axum::extract::State;
+use axum::{
+    body::Body,
+    http::{
+        header::{AUTHORIZATION, WWW_AUTHENTICATE},
+        HeaderMap, Request, StatusCode,
+    },
+    response::{IntoResponse, Response},
+};
+use base64::{engine::general_purpose::STANDARD as BASE64, Engine};
+use dashmap::DashMap;
+use password_hash::phc::PasswordHash;
+use std::{
+    collections::HashMap,
+    fs::File,
+    io::{BufRead, BufReader},
+    sync::Arc,
+    time::{Duration, Instant},
+};
+use tokio::sync::Semaphore;
+use tokio::time::sleep;
+
+fn load_users(path: &str) -> std::io::Result<HashMap<String, String>> {
+    let file = File::open(path)?;
+    let reader = BufReader::new(file);
+
+    let mut users = HashMap::new();
+
+    for line in reader.lines() {
+        let line = line?;
+
+        if let Some((user, hash)) = line.split_once(':') {
+            users.insert(user.to_owned(), hash.to_owned());
+        }
+    }
+
+    Ok(users)
+}
+
+#[derive(Clone)]
+pub struct AuthState {
+    idle_timeout: Duration,
+
+    users: Arc<HashMap<String, String>>, // user -> PHC hash string
+    cache: Arc<DashMap<[u8; 32], Instant>>,
+    /// Bounds concurrent Argon2 verifications (CPU and ~19 MiB each).
+    verify_slots: Arc<Semaphore>,
+}
+
+impl AuthState {
+    pub fn new(path: &str) -> Self {
+        let users = load_users(path).unwrap_or_default();
+
+        Self {
+            idle_timeout: Duration::from_secs(300),
+            cache: Arc::new(DashMap::new()),
+            users: Arc::new(users),
+            verify_slots: Arc::new(Semaphore::new(2)),
+        }
+    }
+}
+
+fn parse_basic(headers: &HeaderMap) -> Option<(String, String)> {
+    let encoded = headers
+        .get(AUTHORIZATION)?
+        .to_str()
+        .ok()?
+        .strip_prefix("Basic ")?;
+    let decoded = String::from_utf8(BASE64.decode(encoded).ok()?).ok()?;
+    let (user, pass) = decoded.split_once(':')?;
+    Some((user.to_owned(), pass.to_owned()))
+}
+
+async fn authenticate(state: &AuthState, headers: &HeaderMap) -> bool {
+    let Some((user, pass)) = parse_basic(headers) else {
+        return false;
+    };
+    let Some(hash) = state.users.get(&user).cloned() else {
+        return false;
+    };
+
+    let key = *blake3::hash(format!("{user}\0{pass}\0{hash}").as_bytes()).as_bytes();
+
+    // Fast path. The shard guard lives only until the end of this `match`;
+    // never call into the map while holding it.
+    let fresh = match state.cache.get_mut(&key) {
+        Some(mut last_seen) if last_seen.elapsed() < state.idle_timeout => {
+            *last_seen = Instant::now(); // slide
+            true
+        }
+        _ => false,
+    };
+    if fresh {
+        return true;
+    }
+
+    // Slow path: Argon2 is deliberately expensive, so keep it off the runtime
+    // threads and bound how many run at once.
+    let Ok(_permit) = state.verify_slots.acquire().await else {
+        return false;
+    };
+
+    let verified = tokio::task::spawn_blocking(move || {
+        PasswordHash::new(&hash)
+            .map(|ph| {
+                Argon2::default()
+                    .verify_password(pass.as_bytes(), &ph)
+                    .is_ok()
+            })
+            .unwrap_or(false)
+    })
+    .await
+    .unwrap_or(false);
+
+    if verified {
+        // Overwrites an expired entry; no separate remove needed.
+        state.cache.insert(key, Instant::now());
+    }
+
+    verified
+}
+
+pub async fn basic_auth(
+    State(state): State<AuthState>,
+    headers: HeaderMap,
+    request: Request<Body>,
+    next: axum::middleware::Next,
+) -> Response {
+    if !authenticate(&state, &headers).await {
+        sleep(Duration::from_millis(1500)).await;
+        return (
+            StatusCode::UNAUTHORIZED,
+            [(WWW_AUTHENTICATE, r#"Basic realm="Restricted""#)],
+            "Unauthorized",
+        )
+            .into_response();
+    }
+
+    next.run(request).await
+}
diff --git a/src/api/debug.rs b/src/api/debug.rs
index 3c24f62..929d665 100644
--- a/src/api/debug.rs
+++ b/src/api/debug.rs
@@ -5,8 +5,8 @@ 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;
+use crate::transport::firewall::FirewallRules;
 
 pub(crate) async fn get_debug_firewall(State(state): State<AppState>) -> Json<FirewallRules> {
     Json(state.firewall.rules())
diff --git a/src/api/mod.rs b/src/api/mod.rs
index 98400c7..ac6f2d0 100644
--- a/src/api/mod.rs
+++ b/src/api/mod.rs
@@ -1,5 +1,6 @@
 //! HTTP API: router, OpenAPI document and shared response helpers.
 
+pub(crate) mod auth;
 mod debug;
 mod peers;
 mod store;
@@ -19,9 +20,9 @@ use tracing::{error, warn};
 use utoipa::OpenApi;
 use utoipa_swagger_ui::SwaggerUi;
 
-use crate::auth::{basic_auth, AuthState};
+use crate::api::auth::{basic_auth, AuthState};
+use crate::document::store::StoreError;
 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 {
diff --git a/src/api/peers.rs b/src/api/peers.rs
index b12e3bf..5308eb9 100644
--- a/src/api/peers.rs
+++ b/src/api/peers.rs
@@ -14,10 +14,10 @@ use tracing::warn;
 use utoipa::ToSchema;
 
 use super::store_status;
-use crate::peer::forgotten_path;
-use crate::replica::Epoch;
+use crate::document::replica::Epoch;
+use crate::document::store;
 use crate::state::{unix_millis, AppState, HelloState, Liveness, PeerMembership, PeerState};
-use crate::store;
+use crate::swarm::peer::forgotten_path;
 
 #[derive(Debug, Serialize, ToSchema)]
 #[serde(rename_all = "lowercase")]
diff --git a/src/api/store.rs b/src/api/store.rs
index e991005..8ea0202 100644
--- a/src/api/store.rs
+++ b/src/api/store.rs
@@ -17,9 +17,9 @@ use tokio::sync::broadcast;
 use tracing::{debug, error, instrument};
 
 use super::{parse_key_path, response_for_value, store_status};
+use crate::document::store::{self, StoreError};
 use crate::state::{AppState, ReadSource};
-use crate::store::{self, StoreError};
-use crate::sync::paths_related;
+use crate::swarm::sync::paths_related;
 
 #[derive(Debug, Default, Deserialize)]
 pub(crate) struct ReadQuery {