diff --git a/gateway/src/gateway.rs b/gateway/src/gateway.rs index 19bf215..ab45581 100644 --- a/gateway/src/gateway.rs +++ b/gateway/src/gateway.rs @@ -86,20 +86,52 @@ pub fn build_gateway(modules: Vec>) -> Result]) -> Vec { + modules + .iter() + .map(|module| ServiceInfo { + name: module.name(), + path: module.path(), + description: module.description(), + enabled: module.enabled(), + }) + .collect() +} + +fn enabled_service_router(modules: &[Arc]) -> Router { + modules + .iter() + .filter(|module| module.enabled()) + .fold(Router::new(), |router, module| { + router.nest(&format!("/{}", module.path()), module.router()) + }) +} + +async fn csrf_token_handler( + session: tower_sessions::Session, +) -> Result, crate::auth::error::AppError> { + let token = crate::csrf::get_or_create_token(&session) + .await + .map_err(|error| { + crate::auth::error::AppError::internal( + "csrf token bootstrap", + std::io::Error::other(error), + ) + })?; + Ok(Json(serde_json::json!({ + "csrf_token": token.as_str(), + "header": crate::csrf::TOKEN_HEADER, + }))) +} + +fn gateway_state( modules: Vec>, mut settings: settings::Settings, proxy_upstream_url: String, db_pool: Option, refresh_token_ttl_secs: i64, access_token_ttl_secs: i64, -) -> Result { +) -> Result<(GatewayState, Router), anyhow::Error> { if refresh_token_ttl_secs <= 0 || access_token_ttl_secs <= 0 { anyhow::bail!("gateway token TTLs must be positive"); } @@ -109,40 +141,51 @@ pub fn build_gateway_with_settings( settings.access_token_ttl_secs = access_token_ttl_secs; let (tx, _rx) = broadcast::channel(settings.sse_broadcast_buffer); - - let service_infos: Vec = modules - .iter() - .map(|module| ServiceInfo { - name: module.name(), - path: module.path(), - description: module.description(), - enabled: module.enabled(), - }) - .collect(); - - let mut service_router: Router = Router::new(); - for module in &modules { - if module.enabled() { - service_router = service_router.nest(&format!("/{}", module.path()), module.router()); - } - } - + let services = service_infos(&modules); + let service_router = enabled_service_router(&modules); let http_client = reqwest::Client::builder() .timeout(Duration::from_secs(5)) .build() .context("failed to build reqwest client")?; - let state = GatewayState { - tx, - services: service_infos, + Ok(( + GatewayState { + tx, + services, + modules, + settings, + proxy_upstream_url: proxy_upstream_url.into(), + db_pool, + refresh_token_ttl_secs, + access_token_ttl_secs, + http_client, + }, + service_router, + )) +} + +/// Compose every `ServiceModule` with pre-loaded settings and runtime state. +/// +/// # Errors +/// Returns an error if TTLs are invalid, the HTTP client cannot be built, or +/// rate-limiter configuration fails. +#[instrument(skip(modules, settings, db_pool))] +pub fn build_gateway_with_settings( + modules: Vec>, + settings: settings::Settings, + proxy_upstream_url: String, + db_pool: Option, + refresh_token_ttl_secs: i64, + access_token_ttl_secs: i64, +) -> Result { + let (state, service_router) = gateway_state( modules, settings, - proxy_upstream_url: proxy_upstream_url.into(), + proxy_upstream_url, db_pool, refresh_token_ttl_secs, access_token_ttl_secs, - http_client, - }; + )?; let (prometheus_layer, metric_handle) = PrometheusMetricLayer::pair(); @@ -222,23 +265,6 @@ pub fn build_gateway_with_settings( let session_router = crate::session::router::().route_layer(csrf_middleware); - async fn csrf_token_handler( - session: tower_sessions::Session, - ) -> Result, crate::auth::error::AppError> { - let token = crate::csrf::get_or_create_token(&session) - .await - .map_err(|e| { - crate::auth::error::AppError::internal( - "csrf token bootstrap", - std::io::Error::other(e), - ) - })?; - Ok(Json(serde_json::json!({ - "csrf_token": token.as_str(), - "header": crate::csrf::TOKEN_HEADER, - }))) - } - let app = Router::new() .route("/auth/csrf", get(csrf_token_handler)) .merge(login_router) diff --git a/live-search/src/app.rs b/live-search/src/app.rs index fdbf091..8a57d72 100644 --- a/live-search/src/app.rs +++ b/live-search/src/app.rs @@ -32,10 +32,6 @@ use crate::styles; /// # Errors /// Returns a server-function error for invalid input, unavailable application /// state, or a database failure. -#[expect( - clippy::unused_async, - reason = "The server body is async only in the SSR build generated by #[server]" -)] #[server(endpoint = "search")] pub async fn search(query: String) -> Result>, ServerFnError> { #[cfg(feature = "ssr")] @@ -120,6 +116,10 @@ pub fn shell(options: LeptosOptions) -> impl IntoView { } #[must_use] +#[expect( + clippy::must_use_candidate, + reason = "Leptos #[component] generates a wrapper that nightly Clippy evaluates separately" +)] #[component] pub fn App() -> impl IntoView { provide_meta_context(); @@ -151,6 +151,10 @@ pub fn App() -> impl IntoView { } #[component] +#[expect( + clippy::empty_enums, + reason = "Leptos #[component] generates an empty typed-builder state enum" +)] fn SearchErrorBoundary(children: Children) -> impl IntoView { view! { , @@ -211,6 +215,10 @@ fn result_row_renderer( } #[must_use] +#[expect( + clippy::must_use_candidate, + reason = "Leptos #[component] generates a wrapper that nightly Clippy evaluates separately" +)] #[component] pub fn SearchPage() -> impl IntoView { let (query, set_query) = signal(String::new()); @@ -325,117 +333,150 @@ struct LiveResult { snippet: Arc, } -#[must_use] -#[component] -pub fn LiveFeedPage() -> impl IntoView { - let results = RwSignal::new(VecDeque::::new()); - let connected = RwSignal::new(false); - - #[cfg(target_arch = "wasm32")] - { - let stop = RwSignal::new(false); - let stop_cleanup = stop; - on_cleanup(move || stop_cleanup.set(true)); - - leptos::task::spawn_local(async move { - use futures::{StreamExt, stream}; - use gloo_net::eventsource::State; - use gloo_timers::future::sleep; - use leptos::logging; - use std::time::Duration; - - loop { - if stop.get() { - logging::log!("SSE live feed stopped"); - return; +#[cfg(target_arch = "wasm32")] +fn apply_sse_event( + event: SseEvent, + results: RwSignal>, + connected: RwSignal, +) { + use leptos::logging; + + match event { + SseEvent::Connected { .. } => connected.set(true), + SseEvent::SearchResult { + title, + url, + snippet, + } => { + results.update(|items| { + if items.len() >= 200 { + items.pop_front(); } + items.push_back(LiveResult { + title, + url, + snippet, + }); + }); + } + SseEvent::StreamLagged { skipped } => { + logging::warn!("SSE stream lagged by {skipped} messages"); + } + } +} - match gloo_net::eventsource::futures::EventSource::new("/api/events") { - Ok(mut event_source) => { - let connected_stream = event_source.subscribe("connected"); - let result_stream = event_source.subscribe("search_result"); - let lagged_stream = event_source.subscribe("stream_lagged"); - - match (connected_stream, result_stream, lagged_stream) { - (Ok(connected_events), Ok(result_events), Ok(lagged_events)) => { - while event_source.state() == State::Connecting && !stop.get() { - sleep(Duration::from_millis(25)).await; - } +#[cfg(target_arch = "wasm32")] +async fn consume_event_source( + event_source: &mut gloo_net::eventsource::futures::EventSource, + results: RwSignal>, + connected: RwSignal, + stop: RwSignal, +) { + use futures::{StreamExt, stream}; + use gloo_net::eventsource::State; + use gloo_timers::future::sleep; + use leptos::logging; + use std::time::Duration; + + let connected_stream = event_source.subscribe("connected"); + let result_stream = event_source.subscribe("search_result"); + let lagged_stream = event_source.subscribe("stream_lagged"); + let (Ok(connected_events), Ok(result_events), Ok(lagged_events)) = + (connected_stream, result_stream, lagged_stream) + else { + logging::error!("Failed to subscribe to named SSE events"); + return; + }; - if stop.get() { - return; - } + while event_source.state() == State::Connecting && !stop.get() { + sleep(Duration::from_millis(25)).await; + } + if stop.get() { + return; + } + if event_source.state() != State::Open { + logging::warn!("SSE connection closed before opening"); + return; + } - if event_source.state() == State::Open { - let data_events = stream::select(result_events, lagged_events); - let mut all_events = - stream::select(connected_events, data_events); - while let Some(result) = all_events.next().await { - if stop.get() { - return; - } - match result { - Ok((_event_type, msg)) => { - let Some(data) = msg.data().as_string() else { - logging::warn!( - "SSE message had non-string data" - ); - continue; - }; - match serde_json::from_str::(&data) { - Ok(SseEvent::Connected { .. }) => { - connected.set(true); - } - Ok(SseEvent::SearchResult { - title, - url, - snippet, - }) => { - results.update(|r| { - if r.len() >= 200 { - r.pop_front(); - } - r.push_back(LiveResult { - title, - url, - snippet, - }); - }); - } - Ok(SseEvent::StreamLagged { skipped }) => { - logging::warn!( - "SSE stream lagged by {skipped} messages" - ); - } - Err(e) => { - logging::warn!( - "Invalid SSE message: {e:?}" - ); - } - } - } - Err(e) => { - logging::warn!("SSE stream error: {e:?}"); - break; - } - } - } - } else { - logging::warn!("SSE connection closed before opening"); - } - } - _ => logging::error!("Failed to subscribe to named SSE events"), - } - connected.set(false); - event_source.close(); - } - Err(e) => logging::warn!("Failed to create EventSource: {e}"), + let data_events = stream::select(result_events, lagged_events); + let mut all_events = stream::select(connected_events, data_events); + while let Some(result) = all_events.next().await { + if stop.get() { + return; + } + match result { + Ok((_event_type, msg)) => { + let Some(data) = msg.data().as_string() else { + logging::warn!("SSE message had non-string data"); + continue; + }; + match serde_json::from_str::(&data) { + Ok(event) => apply_sse_event(event, results, connected), + Err(error) => logging::warn!("Invalid SSE message: {error:?}"), } + } + Err(error) => { + logging::warn!("SSE stream error: {error:?}"); + return; + } + } + } +} + +#[cfg(target_arch = "wasm32")] +async fn run_live_feed( + results: RwSignal>, + connected: RwSignal, + stop: RwSignal, +) { + use gloo_net::eventsource::futures::EventSource; + use gloo_timers::future::sleep; + use leptos::logging; + use std::time::Duration; + + loop { + if stop.get() { + logging::log!("SSE live feed stopped"); + return; + } - sleep(Duration::from_secs(2)).await; + match EventSource::new("/api/events") { + Ok(mut event_source) => { + consume_event_source(&mut event_source, results, connected, stop).await; + connected.set(false); + event_source.close(); } - }); + Err(error) => logging::warn!("Failed to create EventSource: {error}"), + } + + if stop.get() { + return; + } + sleep(Duration::from_secs(2)).await; } +} + +#[cfg(target_arch = "wasm32")] +fn start_live_feed(results: RwSignal>, connected: RwSignal) { + let stop = RwSignal::new(false); + let stop_cleanup = stop; + on_cleanup(move || stop_cleanup.set(true)); + leptos::task::spawn_local(run_live_feed(results, connected, stop)); +} + +#[must_use] +#[expect( + clippy::must_use_candidate, + reason = "Leptos #[component] generates a wrapper that nightly Clippy evaluates separately" +)] +#[component] +pub fn LiveFeedPage() -> impl IntoView { + let results = RwSignal::new(VecDeque::::new()); + let connected = RwSignal::new(false); + + #[cfg(target_arch = "wasm32")] + start_live_feed(results, connected); view! {

"Live Feed"

@@ -465,12 +506,12 @@ pub fn LiveFeedPage() -> impl IntoView { } else { items .iter() - .map(|r| { + .map(|result| { view! {
-

{r.title.clone()}

-

{r.snippet.clone()}

- {r.url.clone()} +

{result.title.clone()}

+

{result.snippet.clone()}

+ {result.url.clone()}
} }) diff --git a/live-search/src/bootstrap.rs b/live-search/src/bootstrap.rs index 4eb2ed2..cd72dd7 100644 --- a/live-search/src/bootstrap.rs +++ b/live-search/src/bootstrap.rs @@ -100,15 +100,33 @@ async fn fallback_handler(uri: Uri) -> impl IntoResponse { (StatusCode::NOT_FOUND, format!("Not found: {uri}")) } -/// Bootstraps all subsystems and starts the HTTP server. -/// -/// # Errors -/// Returns an error if configuration, database setup/migrations, or listener -/// binding fails. -pub async fn run() -> anyhow::Result { - init_tracing(); +fn spawn_pg_listener( + tasks: &mut JoinSet>, + pool: sqlx::PgPool, + tx: broadcast::Sender, + cache: CacheHandle, + shutdown: CancellationToken, +) { + let listener_span = tracing::info_span!("pg_listener"); + tasks.spawn( + async move { + db::run_pg_listener(pool, tx, cache, shutdown).await; + Ok(()) + } + .instrument(listener_span), + ); +} - let cfg = rwf_config::Config::load().context("failed to load workspace config")?; +fn listener_addr(default_port: u16) -> SocketAddr { + let port = std::env::var("PORT") + .or_else(|_| std::env::var("LIVE_SEARCH_PORT")) + .ok() + .and_then(|value| value.parse().ok()) + .unwrap_or(default_port); + SocketAddr::from(([0, 0, 0, 0], port)) +} + +async fn create_database_pool(cfg: &rwf_config::Config) -> anyhow::Result { let database_url = std::env::var("DATABASE_URL") .ok() .unwrap_or_else(|| cfg.live_search.database_url.clone()); @@ -126,15 +144,25 @@ pub async fn run() -> anyhow::Result { .await .context("failed to create database pool")?; - // Every service sharing this database resolves the exact same SQLx - // migration history. Keep migrations on the raw SQLx pool because the - // instrumentation wrapper is for application query execution, not SQLx's - // migration/listener-specific APIs. sqlx::migrate!("../migrations") .run(&raw_pool) .await .context("failed to run database migrations")?; + Ok(raw_pool) +} + +/// Bootstraps all subsystems and starts the HTTP server. +/// +/// # Errors +/// Returns an error if configuration, database setup/migrations, or listener +/// binding fails. +pub async fn run() -> anyhow::Result { + init_tracing(); + + let cfg = rwf_config::Config::load().context("failed to load workspace config")?; + let raw_pool = create_database_pool(&cfg).await?; + let app_pool = state::app_pool_from_raw(raw_pool.clone()); let cache_handle = CacheHandle::default(); @@ -150,16 +178,12 @@ pub async fn run() -> anyhow::Result { let shutdown = CancellationToken::new(); let mut tasks = JoinSet::new(); - let listener_token = shutdown.child_token(); - let pool_for_listener = raw_pool.clone(); - let listener_span = tracing::info_span!("pg_listener"); - let cache_for_listener = cache_handle.clone(); - tasks.spawn( - async move { - db::run_pg_listener(pool_for_listener, tx, cache_for_listener, listener_token).await; - Ok(()) - } - .instrument(listener_span), + spawn_pg_listener( + &mut tasks, + raw_pool.clone(), + tx, + cache_handle.clone(), + shutdown.child_token(), ); let conf = get_configuration(None).context("failed to read Leptos configuration")?; @@ -217,12 +241,7 @@ pub async fn run() -> anyhow::Result { let router: Router<()> = router.with_state(leptos_options); - let port: u16 = std::env::var("PORT") - .or_else(|_| std::env::var("LIVE_SEARCH_PORT")) - .ok() - .and_then(|p| p.parse().ok()) - .unwrap_or(cfg.live_search.port); - let addr = SocketAddr::from(([0, 0, 0, 0], port)); + let addr = listener_addr(cfg.live_search.port); tracing::info!("Live search server listening on {addr}"); let listener = tokio::net::TcpListener::bind(addr)