hephaestus/crates/hephd/src/sync.rs

//! Spoke↔hub op-log sync over HTTP (tech-spec §6.1, §12).
//!
//! The merge engine itself lives in `heph-core` (deterministic, transport-free).
//! This module is the **transport**: a [`router`] the **hub** (server mode)
//! mounts, and [`sync_once`] a **spoke** (`local` + `hub_url`) runs to exchange
//! ops with that hub. Both speak JSON over HTTP with two routes:
//!
//! - `POST /sync/push` — the spoke sends its new ops; the hub merges them.
//! - `GET  /sync/pull?after=<hlc>` — the hub returns ops past the spoke's cursor.
//! - `POST /rpc` — the full daemon API ([`crate::rpc::dispatch`]) over HTTP, for
//!   a no-replica `client`-mode [`crate::remote::RemoteStore`] to proxy against.
//!
//! All routes carry permissive CORS headers and answer the browser preflight
//! (`OPTIONS`), so a browser surface (the `heph-pwa` mobile app) can call `/rpc`
//! cross-origin. When the hub is given a `web_root`, unmatched paths fall back to
//! serving that directory's static files (the PWA shell), so the app can be
//! hosted same-origin straight from the hub.
//!
//! Exchange is **incremental by HLC cursor** (`sync_state`, [`heph_core::SyncCursors`]):
//! each side transfers only the tail it hasn't sent/seen. Merge is idempotent,
//! so a re-pushed op the hub already has is a harmless no-op. When the hub is
//! configured with a verifier ([`crate::auth`]), every route requires a valid
//! OIDC bearer token whose `sub` owns the hub (tech-spec §13); spokes attach
//! that token via the `bearer` argument to [`sync_once`].

use std::path::PathBuf;
use std::sync::{Arc, Mutex};

use anyhow::Result;
use axum::extract::{Query, Request, State};
use axum::http::{header, HeaderValue, Method, StatusCode, Uri};
use axum::middleware::{self, Next};
use axum::response::{IntoResponse, Response as AxumResponse};
use axum::routing::{get, post};
use axum::{Json, Router};
use serde::{Deserialize, Serialize};
use serde_json::Value;

use heph_core::{Op, Store};

use crate::auth::{AuthError, TokenVerifier};
use crate::rpc::{self, Response, RpcError, INTERNAL_ERROR};

/// The shared store a hub serves from — any [`Store`], so a `server` fronts a
/// `LocalStore` and (later modes) could front another backend.
pub type SharedStore = Arc<Mutex<dyn Store + Send>>;

/// What the hub HTTP routes share: the store and (when authentication is
/// configured) the bearer-token verifier.
#[derive(Clone)]
struct HubState {
    store: SharedStore,
    verifier: Option<Arc<dyn TokenVerifier>>,
    /// When set, unmatched paths serve static files from this directory (the
    /// `heph-pwa` shell), so the app can be hosted same-origin from the hub.
    web_root: Option<PathBuf>,
}

/// A batch of ops in flight (push body / pull response).
#[derive(Debug, Serialize, Deserialize)]
pub struct OpsBody {
    /// The ops, applied in HLC order by the receiver.
    pub ops: Vec<Op>,
}

/// What one [`sync_once`] exchange moved.
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
pub struct SyncReport {
    /// Ops received from the hub.
    pub pulled: usize,
    /// Of the pulled ops, how many were newly applied (not already seen).
    pub applied: usize,
    /// Ops sent to the hub.
    pub pushed: usize,
}

/// Run `f` against the locked store on the blocking pool (DB calls never run on
/// an async worker, tech-spec §3).
async fn with_store<T, F>(store: &SharedStore, f: F) -> Result<T>
where
    F: FnOnce(&mut (dyn Store + Send)) -> heph_core::Result<T> + Send + 'static,
    T: Send + 'static,
{
    let store = store.clone();
    let out = tokio::task::spawn_blocking(move || {
        let mut guard = store.lock().expect("store mutex poisoned");
        f(&mut *guard)
    })
    .await?;
    Ok(out?)
}

/// Apply a batch of ops in HLC order, returning how many were newly applied and
/// the highest HLC seen (the new cursor position).
fn apply_batch(
    store: &mut (dyn Store + Send),
    mut ops: Vec<Op>,
) -> heph_core::Result<(usize, Option<String>)> {
    ops.sort_by(|a, b| a.hlc.cmp(&b.hlc));
    let mut applied = 0;
    let mut max_hlc = None;
    for op in &ops {
        if store.apply_op(op)? {
            applied += 1;
        }
        max_hlc = Some(op.hlc.clone());
    }
    Ok((applied, max_hlc))
}

/// The hub's HTTP router (server mode). Mount it on a TCP listener. When
/// `verifier` is `Some`, every route requires a valid OIDC bearer token whose
/// `sub` owns this hub (tech-spec §13); `None` leaves the hub open (local dev).
pub fn router(store: SharedStore, verifier: Option<Arc<dyn TokenVerifier>>) -> Router {
    router_with_web(store, verifier, None)
}

/// [`router`] plus an optional `web_root`: when `Some(dir)`, paths that don't
/// match an API route serve static files from `dir` (the `heph-pwa` shell),
/// with a `index.html` fallback so the single-page app can deep-link. Static
/// files are served without authentication — they are only the app shell; all
/// data still flows through the auth-gated `/rpc` and `/sync/*` routes.
pub fn router_with_web(
    store: SharedStore,
    verifier: Option<Arc<dyn TokenVerifier>>,
    web_root: Option<PathBuf>,
) -> Router {
    let state = HubState {
        store,
        verifier,
        web_root,
    };
    Router::new()
        .route("/sync/pull", get(pull))
        .route("/sync/push", post(push))
        .route("/rpc", post(rpc_call))
        .route_layer(middleware::from_fn_with_state(state.clone(), require_auth))
        // Unauthenticated: the public OIDC params (issuer + client id) a browser
        // client reads to start a PKCE login. Added after the auth `route_layer`
        // so it is NOT gated — the app needs it *before* it has a token.
        .route("/config", get(config))
        // The static shell is unauthenticated and lives behind the API routes.
        .fallback(serve_static)
        // Outermost: stamp CORS headers on every response and short-circuit the
        // browser's `OPTIONS` preflight (before it reaches auth or routing).
        .layer(middleware::from_fn(cors))
        .with_state(state)
}

/// Permissive-CORS middleware. Answers the browser preflight (`OPTIONS`) with a
/// 204 and stamps `Access-Control-*` headers on every response. The hub is a
/// personal endpoint guarded by bearer tokens (not cookies), so a wildcard
/// origin is safe — there are no ambient credentials for `*` to expose.
async fn cors(request: Request, next: Next) -> AxumResponse {
    let is_preflight = request.method() == Method::OPTIONS;
    let mut response = if is_preflight {
        StatusCode::NO_CONTENT.into_response()
    } else {
        next.run(request).await
    };
    let h = response.headers_mut();
    h.insert(
        header::ACCESS_CONTROL_ALLOW_ORIGIN,
        HeaderValue::from_static("*"),
    );
    h.insert(
        header::ACCESS_CONTROL_ALLOW_METHODS,
        HeaderValue::from_static("GET, POST, OPTIONS"),
    );
    h.insert(
        header::ACCESS_CONTROL_ALLOW_HEADERS,
        HeaderValue::from_static("authorization, content-type"),
    );
    h.insert(
        header::ACCESS_CONTROL_MAX_AGE,
        HeaderValue::from_static("86400"),
    );
    response
}

/// Public OIDC parameters for a browser client (the `heph-pwa`) to start a PKCE
/// login: `{ "issuer", "client_id" }`. Unauthenticated — neither value is a
/// secret. Returns an empty object `{}` when the hub runs without OIDC, so the
/// app can detect that and fall back to a manually pasted token.
async fn config(State(state): State<HubState>) -> Json<Value> {
    let body = state
        .verifier
        .as_ref()
        .and_then(|v| v.oidc_config())
        .map(|(issuer, client_id)| serde_json::json!({ "issuer": issuer, "client_id": client_id }))
        .unwrap_or_else(|| serde_json::json!({}));
    Json(body)
}

/// Serve the PWA shell from `web_root` for any non-API path. Returns 404 when no
/// `web_root` is configured. Unknown paths fall back to `index.html` so the SPA
/// can own its own routing. Path traversal (`..`) is rejected.
async fn serve_static(State(state): State<HubState>, uri: Uri) -> AxumResponse {
    let Some(root) = state.web_root.as_ref() else {
        return StatusCode::NOT_FOUND.into_response();
    };
    let rel = uri.path().trim_start_matches('/');
    if rel.split('/').any(|seg| seg == "..") {
        return StatusCode::BAD_REQUEST.into_response();
    }
    let rel = if rel.is_empty() { "index.html" } else { rel };

    let direct = root.join(rel);
    let index = root.join("index.html");
    // File reads run on the blocking pool (tokio's `fs` feature is off, and DB /
    // disk I/O never runs on an async worker, tech-spec §3).
    let read = tokio::task::spawn_blocking(move || {
        match std::fs::read(&direct) {
            Ok(bytes) => Some((content_type(&direct), bytes)),
            // SPA fallback: serve index.html for unknown (extension-less) routes.
            Err(_) => std::fs::read(&index)
                .ok()
                .map(|bytes| ("text/html; charset=utf-8", bytes)),
        }
    })
    .await;
    match read {
        Ok(Some((ctype, bytes))) => ([(header::CONTENT_TYPE, ctype)], bytes).into_response(),
        _ => StatusCode::NOT_FOUND.into_response(),
    }
}

/// Best-effort content type from a file extension (the handful the PWA serves).
fn content_type(path: &std::path::Path) -> &'static str {
    match path.extension().and_then(|e| e.to_str()) {
        Some("html") => "text/html; charset=utf-8",
        Some("js" | "mjs") => "text/javascript; charset=utf-8",
        Some("css") => "text/css; charset=utf-8",
        Some("json" | "webmanifest") => "application/json; charset=utf-8",
        Some("svg") => "image/svg+xml",
        Some("png") => "image/png",
        Some("ico") => "image/x-icon",
        Some("woff2") => "font/woff2",
        _ => "application/octet-stream",
    }
}

/// Reject any request lacking a valid bearer token whose `sub` owns this hub.
/// A no-op when the hub has no verifier configured (open dev mode).
async fn require_auth(
    State(state): State<HubState>,
    request: Request,
    next: Next,
) -> Result<AxumResponse, StatusCode> {
    let Some(verifier) = state.verifier.clone() else {
        return Ok(next.run(request).await); // open: no auth configured
    };

    let Some(token) = bearer_token(&request) else {
        return Err(StatusCode::UNAUTHORIZED);
    };

    // Verification (and the store gate) hit the network / DB — run off the async
    // worker on the blocking pool.
    let claims = tokio::task::spawn_blocking(move || verifier.verify(&token))
        .await
        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
        .map_err(|e| match e {
            AuthError::Provider(_) => StatusCode::SERVICE_UNAVAILABLE,
            _ => StatusCode::UNAUTHORIZED,
        })?;

    // Multi-tenancy seam: resolve the token's identity to the owner it may act
    // as. Today the hub serves one owner, so this is `Some(that owner)` or
    // `None` (→ 403). When the hub becomes multi-owner, `_owner_id` is what each
    // downstream handler scopes its ops to (rather than the store's lone owner).
    let store = state.store.clone();
    let owner = tokio::task::spawn_blocking(move || {
        store
            .lock()
            .expect("store mutex poisoned")
            .resolve_owner(&claims.sub)
    })
    .await
    .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
    .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
    let Some(_owner_id) = owner else {
        return Err(StatusCode::FORBIDDEN);
    };

    Ok(next.run(request).await)
}

/// Extract the `Authorization: Bearer <token>` value, if present.
fn bearer_token(request: &Request) -> Option<String> {
    request
        .headers()
        .get(axum::http::header::AUTHORIZATION)
        .and_then(|v| v.to_str().ok())
        .and_then(|v| v.strip_prefix("Bearer "))
        .map(str::to_string)
}

/// One `POST /rpc` call: a method name + params, mirroring the unix-socket RPC.
#[derive(Debug, Deserialize)]
struct RpcCall {
    method: String,
    #[serde(default)]
    params: Value,
}

/// `POST /rpc` — run one [`rpc::dispatch`] call on the hub's store and return a
/// JSON-RPC-shaped [`Response`] (result xor error). Always HTTP 200; method
/// failures travel in the body so the client can reconstruct the error.
async fn rpc_call(State(state): State<HubState>, Json(call): Json<RpcCall>) -> Json<Response> {
    let store = state.store.clone();
    let dispatched = tokio::task::spawn_blocking(move || {
        let mut guard = store.lock().expect("store mutex poisoned");
        rpc::dispatch(&mut *guard, &call.method, call.params)
    })
    .await;
    let response = match dispatched {
        Ok(Ok(value)) => Response::ok(Value::Null, value),
        Ok(Err(rpc_err)) => Response::failed(Value::Null, rpc_err),
        Err(join_err) => Response::failed(
            Value::Null,
            RpcError {
                code: INTERNAL_ERROR,
                message: format!("dispatch task failed: {join_err}"),
            },
        ),
    };
    Json(response)
}

#[derive(Debug, Deserialize)]
struct PullQuery {
    /// HLC cursor — return ops strictly newer than this (absent ⇒ from the start).
    #[serde(default)]
    after: Option<String>,
}

/// `GET /sync/pull?after=<hlc>` — ops past the caller's cursor, HLC order.
async fn pull(
    State(state): State<HubState>,
    Query(q): Query<PullQuery>,
) -> Result<Json<OpsBody>, StatusCode> {
    let ops = with_store(&state.store, move |s| s.ops_since(q.after.as_deref()))
        .await
        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
    Ok(Json(OpsBody { ops }))
}

/// `POST /sync/push` — merge the caller's ops; reply with how many newly applied.
async fn push(
    State(state): State<HubState>,
    Json(body): Json<OpsBody>,
) -> Result<Json<SyncReport>, StatusCode> {
    let (applied, _max) = with_store(&state.store, move |s| apply_batch(s, body.ops))
        .await
        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
    Ok(Json(SyncReport {
        applied,
        ..Default::default()
    }))
}

/// Exchange ops with `hub_url` once: pull new ops and merge them, then push our
/// new ops. Advances the per-hub cursors so the next call transfers only the
/// tail. `http` is a shared [`reqwest::Client`].
pub async fn sync_once(
    store: SharedStore,
    hub_url: &str,
    http: &reqwest::Client,
    bearer: Option<&str>,
) -> Result<SyncReport> {
    let base = hub_url.trim_end_matches('/');
    let mut report = SyncReport::default();

    let cursors = {
        let hub = hub_url.to_string();
        with_store(&store, move |s| s.sync_state(&hub)).await?
    };

    // --- pull then merge ---
    let mut req = http.get(format!("{base}/sync/pull"));
    if let Some(after) = &cursors.last_pulled_hlc {
        req = req.query(&[("after", after)]);
    }
    if let Some(token) = bearer {
        req = req.bearer_auth(token);
    }
    let pulled: OpsBody = req.send().await?.error_for_status()?.json().await?;
    report.pulled = pulled.ops.len();
    if !pulled.ops.is_empty() {
        let (applied, max_pulled) = with_store(&store, move |s| apply_batch(s, pulled.ops)).await?;
        report.applied = applied;
        if let Some(cursor) = max_pulled {
            let hub = hub_url.to_string();
            with_store(&store, move |s| s.record_sync(&hub, None, Some(&cursor))).await?;
        }
    }

    // --- push our tail ---
    let to_push = {
        let after = cursors.last_pushed_hlc.clone();
        with_store(&store, move |s| s.ops_since(after.as_deref())).await?
    };
    report.pushed = to_push.len();
    if !to_push.is_empty() {
        // `ops_since` returns HLC order, so the last is the new cursor.
        let max_pushed = to_push.last().map(|o| o.hlc.clone());
        let mut req = http
            .post(format!("{base}/sync/push"))
            .json(&OpsBody { ops: to_push });
        if let Some(token) = bearer {
            req = req.bearer_auth(token);
        }
        req.send().await?.error_for_status()?;
        if let Some(cursor) = max_pushed {
            let hub = hub_url.to_string();
            with_store(&store, move |s| s.record_sync(&hub, Some(&cursor), None)).await?;
        }
    }

    Ok(report)
}