decision/crates/ld-server/src/lib.rs

//! libreDecision server.
//!
//! One binary serves the JSON API under `/api`, the compiled Svelte interface,
//! and runs four background workers: the electorate refresher (size of the Ğ1
//! web of trust), the adoption sweeper (with the deadlines of three days of
//! objections and typo fixes), the ballot watcher (votes signed on the Ğ1 chain,
//! confirmed on the indexer, or simulated while the chain is closed — ADR 0012)
//! and the publisher (adopted texts pushed to the git
//! repository that client applications read).

pub mod access;
pub mod adoption;
pub mod auth;
pub mod ballots;
pub mod clock;
pub mod config;
pub mod deadlines;
pub mod doc_title;
pub mod docs;
pub mod error;
pub mod indexer;
pub mod inertia_amend;
pub mod messages;
pub mod module_amend;
pub mod modules;
pub mod objections;
pub mod order;
pub mod overview;
pub mod participation;
pub mod proposals;
pub mod publish;
pub mod quorum;
pub mod quorum_amend;
pub mod seed;
pub mod seed_amend;
pub mod seed_proposal_amend;
pub mod simulation;
pub mod team;
pub mod typos;
pub mod util;
pub mod web;
pub mod wot;

use axum::Router;
use axum::extract::{Request, State};
use axum::http::{Method, StatusCode};
use axum::middleware::{self, Next};
use axum::response::{IntoResponse, Response};
use axum::routing::{delete, get, post, put};
use sqlx::SqlitePool;
use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions};
use std::str::FromStr;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;
use tokio::sync::Notify;

pub use config::Config;

#[derive(Clone)]
pub struct AppState {
    pub db: SqlitePool,
    pub cfg: Arc<Config>,
    pub http: reqwest::Client,
    /// Wakes the publisher as soon as an adoption is recorded.
    pub publish_wake: Arc<Notify>,
    /// What is coming in the tools of each group (`seeds/upcoming/`), read at start.
    pub upcoming: Arc<overview::Upcoming>,
    /// The clock of the deadlines of three days (objections, typo fixes); tests move it.
    pub clock: clock::Clock,
    /// Tests only, with the dev login: the chain of this server closed for a while
    /// (`POST /api/dev/vote-chain`), to try the simulation (ADR 0012).
    pub chain_closed: Arc<AtomicBool>,
}

impl AppState {
    pub async fn new(cfg: Config, db: SqlitePool) -> anyhow::Result<Self> {
        let http = reqwest::Client::builder()
            .timeout(Duration::from_secs(15))
            .user_agent(concat!("libreDecision/", env!("CARGO_PKG_VERSION")))
            .build()?;
        let upcoming = Arc::new(overview::Upcoming::load(&cfg.seed_dir).await?);
        Ok(Self {
            db,
            cfg: Arc::new(cfg),
            http,
            publish_wake: Arc::new(Notify::new()),
            upcoming,
            clock: clock::Clock::default(),
            chain_closed: Arc::new(AtomicBool::new(false)),
        })
    }

    /// The vote on the chain, as it is now: `None` while it is closed (no vote
    /// account, or closed for a test) — the vote is then simulated (ADR 0012).
    pub fn chain(&self) -> Option<&config::VoteChain> {
        self.cfg.vote.as_ref().filter(|_| !self.chain_closed.load(Ordering::Relaxed))
    }

    /// What each stage of a vote weighs (ADR 0012).
    pub fn vote_rules(&self) -> ld_engine::buffer::Rules {
        ld_engine::buffer::Rules { simulated_adopts: self.cfg.simulated_adopts }
    }
}

/// Open (or create) the database and apply migrations.
pub async fn open_db(url: &str) -> anyhow::Result<SqlitePool> {
    let options = SqliteConnectOptions::from_str(url)?
        .create_if_missing(true)
        .journal_mode(SqliteJournalMode::Wal)
        .foreign_keys(true)
        .busy_timeout(Duration::from_secs(5));
    let max = if url.contains(":memory:") { 1 } else { 8 };
    let pool = SqlitePoolOptions::new().max_connections(max).connect_with(options).await?;
    // A previous image (rollback) must still start on a database a newer one has
    // migrated: migrations are additive, the unknown ones are tolerated.
    let mut migrator = sqlx::migrate!("./migrations");
    migrator.set_ignore_missing(true);
    migrator.run(&pool).await?;
    Ok(pool)
}

/// Mutating API calls must carry this header: a cross-site form cannot set it,
/// and no CORS is ever granted, so it closes CSRF together with SameSite=Lax.
pub const CSRF_HEADER: &str = "x-requested-with";
pub const CSRF_VALUE: &str = "libredecision";

async fn csrf_guard(req: Request, next: Next) -> Response {
    let safe = matches!(*req.method(), Method::GET | Method::HEAD | Method::OPTIONS);
    let has_header = req.headers().get(CSRF_HEADER).and_then(|v| v.to_str().ok()) == Some(CSRF_VALUE);
    if safe || has_header {
        next.run(req).await
    } else {
        (StatusCode::FORBIDDEN, axum::Json(serde_json::json!({ "error": "csrf" }))).into_response()
    }
}

async fn health(State(state): State<AppState>) -> impl IntoResponse {
    let db_ok = sqlx::query_scalar::<_, i64>("SELECT 1").fetch_one(&state.db).await.is_ok();
    axum::Json(serde_json::json!({ "ok": db_ok, "build": state.cfg.build }))
}

pub fn api_router() -> Router<AppState> {
    Router::new()
        .route("/health", get(health))
        .route("/auth/config", get(auth::config))
        .route("/auth/g1/start", get(auth::g1_start))
        .route("/auth/g1/callback", get(auth::g1_callback))
        .route("/auth/invitation", post(auth::accept_invitation))
        .route("/auth/team", post(team::join))
        .route("/auth/team/peek", post(team::peek))
        .route("/auth/dev", post(auth::dev_login))
        .route("/auth/logout", post(auth::logout))
        .route("/me", get(auth::me))
        .route("/groups", get(docs::groups))
        .route("/groups/{slug}", get(overview::group))
        .route("/groups/{slug}/decisions", get(overview::decisions))
        .route("/docs/{id}", get(docs::document))
        .route("/docs/{id}/preview", get(docs::preview))
        .route("/docs/{id}/assist", post(proposals::assist))
        .route("/docs/{id}/proposals", post(proposals::create))
        .route("/docs/{id}/votes", post(ballots::vote_many))
        .route("/docs/{id}/ballots", post(ballots::create))
        .route("/ballots", get(ballots::mine))
        .route("/ballots/{id}", get(ballots::status).delete(ballots::drop))
        .route("/ballots/{id}/receipt", get(ballots::receipt))
        .route("/proposals/{id}", get(proposals::detail).patch(proposals::edit))
        .route("/proposals/{id}/submit", post(proposals::submit))
        .route("/proposals/{id}/withdraw", post(proposals::withdraw))
        .route("/proposals/{id}/split", post(proposals::split))
        .route("/proposals/{id}/vote", post(proposals::vote))
        .route("/proposals/{id}/objections", post(objections::raise))
        .route("/proposals/{id}/translations/{lang}", put(proposals::translate))
        .route("/objections/{id}/lift", post(objections::lift))
        .route("/objections/{id}/reiterate", post(objections::reiterate))
        .route("/typos/{id}/validate", post(typos::validate))
        .route("/typos/{id}/refuse", post(typos::refuse))
        .route("/messages", get(messages::list))
        .route("/messages/unread", get(messages::unread))
        .route("/messages/read", post(messages::read_all))
        .route("/messages/{id}/read", post(messages::read))
        .route("/dev/clock", post(deadlines::dev_clock))
        .route("/dev/vote-chain", post(ballots::dev_chain))
        .route("/units/{id}/inertia", post(proposals::propose_inertia))
        .route("/units/{id}/quorum", put(quorum::set))
        .route("/units/{id}/order", post(order::propose))
        .route("/units/{id}/title", post(doc_title::propose))
        .route("/feedback", post(access::send_feedback))
        .route("/access", get(access::overview))
        .route("/access/groups/{slug}/members", post(access::grant))
        .route("/access/groups/{slug}/members/{identity}", delete(access::revoke))
        .route("/access/invitations", post(access::invite))
        .route("/access/invitations/{hash}", delete(access::cancel_invitation))
        .route("/access/publication/retry", post(access::publish_now))
        .route("/access/team-links", post(team::create))
        .route("/access/team-links/{hash}", delete(team::revoke))
        .route("/access/identities/{id}/admin", post(access::set_admin))
        .route("/access/identities/{id}/sessions", delete(access::close_sessions))
        .route("/access/feedback", get(access::feedback_list))
        .route("/access/feedback/{id}", post(access::feedback_status))
        .layer(middleware::from_fn(csrf_guard))
}

pub fn app(state: AppState) -> Router {
    use axum::http::{HeaderName, HeaderValue, header};
    use tower_http::set_header::SetResponseHeaderLayer;
    // Neighbouring sites under *.librodrome.org are same-site: none may frame us.
    let security: [(HeaderName, &'static str); 4] = [
        (header::X_FRAME_OPTIONS, "DENY"),
        (header::CONTENT_SECURITY_POLICY, "frame-ancestors 'none'"),
        (header::X_CONTENT_TYPE_OPTIONS, "nosniff"),
        (header::REFERRER_POLICY, "same-origin"),
    ];
    let mut router = Router::new().nest("/api", api_router()).fallback_service(web::service(&state.cfg.web_dir));
    for (name, value) in security {
        router = router.layer(SetResponseHeaderLayer::if_not_present(name, HeaderValue::from_static(value)));
    }
    router
        .layer(tower_http::limit::RequestBodyLimitLayer::new(256 * 1024))
        .layer(tower_http::compression::CompressionLayer::new())
        .layer(tower_http::trace::TraceLayer::new_for_http())
        .with_state(state)
}

/// Ctrl-C, or SIGTERM from `docker stop`.
async fn shutdown() {
    let ctrl_c = async {
        let _ = tokio::signal::ctrl_c().await;
    };
    #[cfg(unix)]
    let term = async {
        match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) {
            Ok(mut s) => {
                s.recv().await;
            }
            Err(_) => std::future::pending::<()>().await,
        }
    };
    #[cfg(not(unix))]
    let term = std::future::pending::<()>();
    tokio::select! {
        () = ctrl_c => {}
        () = term => {}
    }
}

pub async fn run() -> anyhow::Result<()> {
    tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::try_from_default_env()
                .unwrap_or_else(|_| "info,sqlx=warn,tower_http=warn".into()),
        )
        .with_ansi(std::io::IsTerminal::is_terminal(&std::io::stdout()))
        .init();
    let cfg = Config::from_env();
    tokio::fs::create_dir_all(&cfg.data_dir).await?;
    let db_url = format!("sqlite://{}", cfg.data_dir.join("libredecision.db").display());
    let db = open_db(&db_url).await?;
    let state = AppState::new(cfg, db).await?;

    seed::ensure(&state).await?;
    // A measure missed leaves the reference in place; the sweeper takes it again.
    if let Err(e) = participation::refresh(&state, false).await {
        tracing::warn!(error = %e, "usual participation");
    }
    auth::bootstrap(&state).await?;
    // A deadline passed while the server was stopped is dealt with now.
    match deadlines::sweep(&state).await {
        Ok(s) if s != deadlines::Swept::default() => tracing::info!(?s, "deadlines passed while stopped"),
        Ok(_) => {}
        Err(e) => tracing::warn!(error = %e, "deadlines"),
    }
    if let Err(e) = publish::ensure_key(&state).await {
        tracing::warn!(error = %e, "publication key");
    }
    publish::resync(&state).await?;
    tokio::spawn(wot::refresher(state.clone()));
    tokio::spawn(publish::worker(state.clone()));
    tokio::spawn(adoption::sweeper(state.clone()));
    tokio::spawn(ballots::watcher(state.clone()));

    let listener = tokio::net::TcpListener::bind(&state.cfg.bind).await?;
    tracing::info!(bind = %state.cfg.bind, build = %state.cfg.build, "libreDecision listening");
    axum::serve(listener, app(state).into_make_service_with_connect_info::<std::net::SocketAddr>())
        .with_graceful_shutdown(shutdown())
        .await?;
    Ok(())
}