portal/src/main.rs

/// One binary, three jobs: `portal` (or `portal serve`) runs a site;
/// `portal lint` checks a content repo the way the site reads it;
/// `portal plan`, `build`, `report` and the rest train one. The lint
/// and the trainer are the same code as the server, at the same
/// version, so they never disagree about what content means.
#[cfg(feature = "ssr")]
#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let argv: Vec<String> = std::env::args().skip(1).collect();
    match argv.first().map(String::as_str) {
        Some("lint") => return portal::lint::run(argv[1..].to_vec()).await,
        Some("plan" | "build" | "answer" | "rate" | "prefer" | "score" | "simulate" | "compare" | "report" | "generate") => {
            // The trainer runs models through shell commands and waits
            // on them; off the runtime's threads, as a CLI should.
            return tokio::task::spawn_blocking(move || portal::trainer::run(argv)).await?;
        }
        Some("--version" | "-V") => {
            println!("portal v{}", env!("CARGO_PKG_VERSION"));
            return Ok(());
        }
        Some("serve") | None => {}
        Some(other) => anyhow::bail!("portal: unknown command {other:?}; one of serve, lint, plan, build, answer, rate, prefer, score, simulate, compare, report, generate"),
    }
    serve().await
}

#[cfg(feature = "ssr")]
async fn serve() -> anyhow::Result<()> {
    use axum::{body::Body, extract::State, http::Request, response::IntoResponse, routing::{any, get, post}, Router};
    use leptos::prelude::*;
    use leptos_axum::{generate_route_list, LeptosRoutes};
    use portal::app::{shell, App};
    use portal::content;
    use portal::server::{oidc, AppState};
    use portal::upload::{self, Garage};
    use std::sync::Arc;
    use tower_http::services::{ServeDir, ServeFile};
    use tower_sessions::SessionManagerLayer;

    dotenvy::dotenv().ok();
    tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::try_from_default_env()
                .unwrap_or_else(|_| "info,portal=debug".into()),
        )
        .init();

    let conf = get_configuration(None)?;
    let leptos_options = conf.leptos_options;
    let addr = leptos_options.site_addr;
    let routes = generate_route_list(App);

    let content_repo = std::env::var("CONTENT_REPO")
        .unwrap_or_else(|_| "https://project.uhhm.no/uhhm/questions".to_string());
    let content_branch = std::env::var("CONTENT_BRANCH").unwrap_or_else(|_| "main".to_string());
    let gitea_base = content::gitea_api_base(&content_repo)?;
    let content_raw_base = content::gitea_raw_base(&content_repo)?;
    let aggregates = content::load_aggregates_from_gitea(&content_repo, &content_branch).await?;
    let questions = content::load_questions_from_gitea(&content_repo, &content_branch, "questions").await?;
    content::validate_questions(&questions, &aggregates)?;
    let site = content::load_site_from_gitea(&content_repo, &content_branch).await?;
    let storylines = content::same_arity(questions.arity(), aggregates.arity(), site.arity())?;
    tracing::info!(count = questions.len(), storylines, repo = %content_repo, branch = %content_branch, "loaded content");
    let access = Arc::new(arc_swap::ArcSwap::from_pointee(portal::access::Policy::from_content(&questions, &aggregates)));
    let questions = Arc::new(arc_swap::ArcSwap::from_pointee(questions));
    let aggregates = Arc::new(arc_swap::ArcSwap::from_pointee(aggregates));
    let site = Arc::new(arc_swap::ArcSwap::from_pointee(site));

    let nats_url =
        std::env::var("NATS_URL").unwrap_or_else(|_| "nats://127.0.0.1:4222".to_string());
    tracing::info!(%nats_url, "connecting to NATS");
    let parsed = url::Url::parse(&nats_url)?;
    let mut nats_opts = async_nats::ConnectOptions::new();
    if !parsed.username().is_empty() {
        nats_opts = nats_opts.user_and_password(
            parsed.username().to_string(),
            parsed.password().unwrap_or_default().to_string(),
        );
    }
    let nats = nats_opts.connect(&nats_url).await?;

    // No fixed, pre-declared list of buckets to initialize here - content
    // names whatever bucket an alternative's `record_as` should write
    // into, and `answers::store_answer` creates it on first use.
    let jetstream = async_nats::jetstream::new(nats.clone());
    portal::events::store::ensure_stream(&jetstream).await?;
    // Outgoing mail waits here for gdo; a mail is never lost to gdo
    // being down, and never sent by portal itself.
    // Best-effort: a stream that already owns these subjects under an
    // older name (a rename mid-upgrade) must not keep a site from
    // booting; mail waits until the old stream is removed.
    if let Err(e) = portal::mail::ensure_stream(&jetstream).await {
        tracing::warn!("the mail stream could not be created: {e} - outgoing mail is off until a stream owns portal.mail.>");
    }

    let oidc_state = Arc::new(oidc::Oidc::from_env().await?);

    let garage = Garage::from_env();
    if garage.is_none() {
        tracing::warn!("GARAGE_* env vars not set - file upload fields will fail closed");
    }

    tokio::spawn(content::watch_for_reload(
        nats.clone(),
        content_repo,
        content_branch.clone(),
        "questions".to_string(),
        questions.clone(),
        aggregates.clone(),
        site.clone(),
        access.clone(),
    ));
    let state = AppState {
        leptos_options: leptos_options.clone(),
        nats,
        jetstream,
        questions,
        aggregates,
        site,
        access,
        gitea_base,
        content_raw_base,
        content_branch: content_branch.clone(),
        oidc: oidc_state,
        garage,
    };

    // The same decisions, answerable by other apps over NATS, for a
    // person named by phone number or address as well as by groups.
    tokio::spawn(portal::access::serve(state.nats.clone(), site_name_for_subjects(), state.access.clone(), Some(state.clone())));
    // Announced pages: keep their desk records current, close ended
    // windows (see announce.rs).
    tokio::spawn(portal::announce::run(state.clone()));
    tokio::spawn(portal::deadlines::run(state.clone()));
    tokio::spawn(portal::deadlines::run_leaving(state.clone()));
    // The accounts the content mandates: the first person in, and
    // everyone a page names as responsible, at start and whenever the
    // content changes (seed.rs).
    tokio::spawn(portal::seed::run(state.clone()));
    // Invites whose link ran out, or whose person is in (people.rs).
    tokio::spawn(portal::people::run(state.clone()));

    // Replies to case mail, matched to their record by whatever reads
    // the mailbox (uhhm/gdo over JMAP) and turned into notes here. A
    // site with no mailbox simply never sees a message on the subject.
    tokio::spawn(portal::mail::receive(state.clone()));

    // Dev-friendly defaults: in-memory sessions, secure cookies only when
    // COOKIE_SECURE=true (set it behind TLS in production) - same
    // defaults cnats uses. SameSite=Lax (tower-sessions defaults to
    // Strict) because the OIDC callback is, by definition, a top-level
    // GET arriving via a redirect *from* the IdP's origin - Strict
    // withholds the cookie on exactly that request, breaking login
    // whenever the app and Kanidm aren't under the same registrable
    // domain (e.g. this app on 127.0.0.1 during local dev vs. Kanidm's
    // real domain). Lax is the standard, correct setting for this
    // pattern, not a workaround.
    let cookie_secure = std::env::var("COOKIE_SECURE")
        .map(|v| v == "true" || v == "1")
        .unwrap_or(false);
    // Sessions live in the same JetStream as everything else, so a
    // deploy no longer signs out every desk on the site it deploys.
    let session_store = portal::sessions::NatsStore::open(&state.jetstream).await?;
    let session_layer = SessionManagerLayer::new(session_store)
        .with_secure(cookie_secure)
        .with_same_site(tower_sessions::cookie::SameSite::Lax)
        .with_name("portal_session");

    // What a person does costs a little; what a script does costs a
    // lot. More than a hundred requests in ten seconds is a script: a
    // burst of 100, refilled at ten a second. Someone reloading a page
    // again and again stays well inside it, the more so since the
    // files a page is made of (stylesheet, scripts, fonts, pictures)
    // are not counted at all - see `assets` below. Per caller IP, in
    // memory, no store: this is a fence, not an accounting system, and
    // the fence that matters for uploads is the policy check on the
    // route itself.
    //
    // PORTAL_RATE_PER_SECOND / PORTAL_RATE_BURST retune it per site;
    // 0 for either turns it off, for an instance behind something that
    // already does this.
    let rate_per_second: u64 = std::env::var("PORTAL_RATE_PER_SECOND")
        .ok()
        .and_then(|v| v.parse().ok())
        .unwrap_or(10);
    let rate_burst: u32 = std::env::var("PORTAL_RATE_BURST")
        .ok()
        .and_then(|v| v.parse().ok())
        .unwrap_or(100);
    let governor = (rate_per_second > 0 && rate_burst > 0).then(|| {
        // SmartIpKeyExtractor reads X-Forwarded-For/X-Real-IP, which is
        // what every one of these instances is behind; it falls back to
        // the socket address when neither is set.
        let conf = tower_governor::governor::GovernorConfigBuilder::default()
            .per_millisecond((1000 / rate_per_second).max(1))
            .burst_size(rate_burst)
            .key_extractor(tower_governor::key_extractor::SmartIpKeyExtractor)
            .finish()
            .expect("a rate limit of at least one per second and a burst of at least one");
        std::sync::Arc::new(conf)
    });
    match &governor {
        Some(_) => tracing::info!(rate_per_second, rate_burst, "rate limiting per caller IP"),
        None => tracing::warn!("rate limiting is off (PORTAL_RATE_PER_SECOND or PORTAL_RATE_BURST is 0)"),
    }

    async fn server_fn_handler(
        State(state): State<AppState>,
        request: Request<Body>,
    ) -> impl IntoResponse {
        leptos_axum::handle_server_fns_with_context(
            move || provide_context(state.clone()),
            request,
        )
        .await
    }

    // The app's own Leptos route list includes a root-level wildcard
    // (`/*any`, since a question's `id` doubles as its URL path), which
    // would otherwise swallow every request - including these static
    // assets - before `.fallback()` ever got a chance to serve them.
    // Registering them explicitly here lets axum's route matching pick
    // the more specific match over the wildcard.
    let pkg_dir = format!(
        "{}/{}",
        leptos_options.site_root, leptos_options.site_pkg_dir
    );
    let favicon_light_path = format!("{}/favicon-light.svg", leptos_options.site_root);
    let favicon_dark_path = format!("{}/favicon-dark.svg", leptos_options.site_root);
    let wordmark_path = format!("{}/wordmark.svg", leptos_options.site_root);
    let swiper_path = format!("{}/swiper-element-bundle.min.js", leptos_options.site_root);
    // Vendored, not loaded from a CDN: a map must not be the one thing
    // on the page that phones a stranger.
    let maplibre_js = format!("{}/maplibre-gl.js", leptos_options.site_root);
    let maplibre_css = format!("{}/maplibre-gl.css", leptos_options.site_root);
    let fonts_dir = format!("{}/fonts", leptos_options.site_root);
    let page_motion = format!("{}/page-motion.js", leptos_options.site_root);

    // The files a page is made of, from this disk: the stylesheet, the
    // scripts, the fonts. Not counted by the rate limit - they cost
    // next to nothing, and refusing one is what turns a reloaded page
    // into an unstyled mess - and no session is loaded for them.
    // Anything that makes portal ask someone else (content pictures
    // from Gitea, map tiles) stays below, counted.
    let assets: Router<AppState> = Router::new()
        // no-cache = "revalidate before reuse", not "don't cache":
        // pkg files keep the same names across releases (portal.js,
        // portal.wasm), and without this browsers heuristically cache
        // them - a stale wasm from the previous release then talks to
        // a server whose server-fn wire format has moved on and every
        // page renders as its error branch. A 304 per load is the
        // price of never shipping that skew again.
        .nest_service(
            "/pkg",
            tower::ServiceBuilder::new()
                .layer(tower_http::set_header::SetResponseHeaderLayer::overriding(
                    axum::http::header::CACHE_CONTROL,
                    axum::http::HeaderValue::from_static("no-cache"),
                ))
                .service(ServeDir::new(pkg_dir)),
        )
        .nest_service("/fonts", ServeDir::new(fonts_dir))
        .route_service("/favicon-light.svg", ServeFile::new(favicon_light_path))
        .route_service("/favicon-dark.svg", ServeFile::new(favicon_dark_path))
        .route_service("/wordmark.svg", ServeFile::new(wordmark_path))
        .route_service(
            "/swiper-element-bundle.min.js",
            ServeFile::new(swiper_path),
        )
        // Same name every release, so revalidated like /pkg.
        .route_service(
            "/page-motion.js",
            tower::ServiceBuilder::new()
                .layer(tower_http::set_header::SetResponseHeaderLayer::overriding(
                    axum::http::header::CACHE_CONTROL,
                    axum::http::HeaderValue::from_static("no-cache"),
                ))
                .service(ServeFile::new(page_motion)),
        )
        .route_service("/maplibre-gl.js", ServeFile::new(maplibre_js))
        .route_service("/maplibre-gl.css", ServeFile::new(maplibre_css));

    let app = Router::new()
        .route("/auth/login", get(oidc::login))
        .route("/auth/callback", get(oidc::callback))
        .route("/auth/logout", get(oidc::logout))
        .route("/api/{*fn_name}", any(server_fn_handler))
        // axum caps request bodies at 2 MB by default, which had made the
        // 20 MB upload limit unreachable; raise the fence on this route only,
        // with room for the multipart framing.
        .route(
            "/upload",
            post(upload::upload).layer(axum::extract::DefaultBodyLimit::max(upload::MAX_UPLOAD_BYTES + 64 * 1024)),
        )
        .route(
            "/attachment/{bucket}/{record}/{field}",
            get(upload::attachment),
        )
        .route("/gitea-repo", get(content::gitea_repo_handler))
        // What the storylines did, as counts (report.rs).
        .route("/report", get(portal::report::handler))
        // Content-shipped assets (hero module etc.), same-origin.
        .route("/site/{*path}", get(content::site_asset_handler))
        .route("/automation/kv/{bucket}", get(content::automation_kv_handler))
        // Maps, proxied: the browser asks this origin for every tile
        // and every still, so nothing third-party loads in a page and
        // the token stays here (see maps.rs).
        .route("/map/style.json", get(portal::maps::style))
        .route("/map/tiles/{z}/{x}/{y}", get(portal::maps::tile))
        .route("/map/still/{place}/{size}", get(portal::maps::still))
        .leptos_routes_with_context(
            &state,
            routes,
            {
                let state = state.clone();
                move || provide_context(state.clone())
            },
            {
                let leptos_options = leptos_options.clone();
                move || shell(leptos_options.clone())
            },
        )
        .fallback(leptos_axum::file_and_error_handler::<AppState, _>(shell))
        .layer(session_layer)
        // Which storyline this visitor reads, before anything renders.
        .layer(axum::middleware::from_fn_with_state(state.clone(), portal::server::storyline::assign));
    // Outermost, so a caller who is over the limit is turned away
    // before a session is loaded or a handler is entered.
    let app = match governor {
        Some(conf) => app.layer(tower_governor::GovernorLayer::new(conf)),
        None => app,
    };
    let app = assets.merge(app).with_state(state);

    tracing::info!("listening on http://{addr}");
    let listener = tokio::net::TcpListener::bind(&addr).await?;
    // With connect info, not without: the rate limiter reads the
    // caller from `X-Forwarded-For`/`X-Real-IP` and falls back to the
    // socket, and without this there is no socket to fall back to. A
    // request that arrives without those headers - a health check on
    // localhost, anything not through Caddy - then has no key at all,
    // and the limiter answers 500 to every one of them.
    axum::serve(
        listener,
        app.into_make_service_with_connect_info::<std::net::SocketAddr>(),
    )
    .await?;
    Ok(())
}

#[cfg(not(feature = "ssr"))]
fn main() {
    // The browser build is a cdylib; this stub only exists so `cargo check`
    // without features still succeeds.
}

#[cfg(feature = "ssr")]
/// The site's name in NATS subjects (`portal.access.<site>.may`):
/// `SITE_NAME`, folded to what a subject token allows.
fn site_name_for_subjects() -> String {
    let raw = std::env::var("SITE_NAME").unwrap_or_else(|_| "portal".to_string());
    let folded: String = raw
        .chars()
        .map(|c| if c.is_ascii_alphanumeric() || c == '_' || c == '-' { c.to_ascii_lowercase() } else { '_' })
        .collect();
    if folded.is_empty() { "portal".to_string() } else { folded }
}