Skip to main content

max / makenotwork

35.3 KB · 824 lines History Blame Raw
1 //! Application entry point, tracing, config, pool, then hands off to `build_app`.
2
3 use axum::http::Request;
4 use sqlx::ConnectOptions;
5 use sqlx::postgres::PgPoolOptions;
6 use std::time::Duration;
7 use tower_http::request_id::{MakeRequestUuid, PropagateRequestIdLayer, SetRequestIdLayer};
8 use tower_http::trace::TraceLayer;
9 use tower_sessions::cookie::SameSite;
10 use tower_sessions::cookie::time::Duration as CookieDuration;
11 use tower_sessions::{Expiry, SessionManagerLayer};
12 use tower_sessions_sqlx_store::PostgresStore;
13 use tracing_subscriber::{Layer, layer::SubscriberExt, util::SubscriberInitExt};
14
15 use docengine::{DocLoader, DocLoaderConfig};
16 use makenotwork::config::Config;
17 use makenotwork::constants;
18 use makenotwork::email::{EmailClient, EmailConfig};
19 use makenotwork::payments::StripeClient;
20 use makenotwork::scanning::ScanPipeline;
21 use makenotwork::storage::S3Client;
22 use makenotwork::{AppState, AppStorage, build_app};
23 use mnw_assumptions::Assumptions;
24 use webauthn_rs::WebauthnBuilder;
25
26 #[tokio::main]
27 async fn main() {
28 dotenvy::dotenv().ok();
29
30 // JSON in release, human-readable in dev
31 tracing_subscriber::registry()
32 .with(
33 tracing_subscriber::EnvFilter::try_from_default_env()
34 .unwrap_or_else(|_| "makenotwork=debug,tower_http=debug,sqlx=info".into()),
35 )
36 .with(if cfg!(debug_assertions) {
37 tracing_subscriber::fmt::layer().boxed()
38 } else {
39 tracing_subscriber::fmt::layer().json().boxed()
40 })
41 .init();
42
43 // Sando boot-smoke gate spawns the binary with SANDO_BOOT_SMOKE=1 to
44 // verify it loads + links + survives. The real init loads
45 // assumptions.toml and other runtime files that don't exist in the
46 // sando build workspace. Short-circuit to a tiny axum server that
47 // proves the binary, tokio runtime, axum, and TCP bind all work,
48 // then idles until the gate kills it.
49 if std::env::var("SANDO_BOOT_SMOKE").is_ok() {
50 // Sando passes SANDO_BOOT_SMOKE_PORT so the gate can actually probe
51 // GET /health (readiness), not just check the process stays up. A bad
52 // value or unset falls back to an ephemeral port (liveness-only, the
53 // historical behavior) rather than failing the smoke for a config typo.
54 let port = std::env::var("SANDO_BOOT_SMOKE_PORT")
55 .ok()
56 .and_then(|p| p.parse::<u16>().ok())
57 .unwrap_or(0);
58 tracing::info!(port, "SANDO_BOOT_SMOKE=1; running minimal smoke server");
59 let app = axum::Router::new().route("/health", axum::routing::get(|| async { "ok" }));
60 let listener = tokio::net::TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, port))
61 .await
62 .expect("smoke: bind 127.0.0.1");
63 axum::serve(listener, app).await.expect("smoke: serve");
64 return;
65 }
66
67 // Config-only validation mode. Loads Config::from_env() with the process
68 // environment and exits, no DB connection, no migrations, no socket bind,
69 // so a required var missing on a target (e.g. CDN_BASE_URL) is caught
70 // *before* a deploy swaps the `current` symlink and restarts into a crash
71 // loop. Sando's pre-swap config-drift guard runs `MNW_CHECK_CONFIG=1
72 // <binary>` on the target with its env sourced; also handy by hand. Prints a
73 // stable sentinel line so a caller can distinguish a genuine config error
74 // (exit 1) from a binary too old to support this mode (which would ignore
75 // the var and fall through to normal startup).
76 if std::env::var("MNW_CHECK_CONFIG").is_ok() {
77 match Config::from_env() {
78 Ok(_) => {
79 println!("MNW_CONFIG_CHECK: ok");
80 std::process::exit(0);
81 }
82 Err(e) => {
83 eprintln!("MNW_CONFIG_CHECK: error: {e}");
84 std::process::exit(1);
85 }
86 }
87 }
88
89 // Docs integrity check mode. Loads the doc corpus with the exact production
90 // config and reports broken internal links, then exits: no DB, no socket,
91 // so it is cheap. Sando's `code_smoke` gate runs `MNW_CHECK_DOCS=1 <binary>`
92 // as its first step, before the throwaway-DB boot, so a rotted internal link
93 // fails the pipeline early rather than serving a live 404 in prod. Broken
94 // links fail the check; slug collisions are reported but do not fail it
95 // (pre-existing collisions shouldn't red the pipeline). Prints a stable
96 // sentinel line, like MNW_CHECK_CONFIG.
97 if std::env::var("MNW_CHECK_DOCS").is_ok() {
98 let assumptions = match load_assumptions() {
99 Ok(a) => a,
100 Err(e) => {
101 eprintln!("MNW_CHECK_DOCS: error: {e}");
102 std::process::exit(1);
103 }
104 };
105 let docs = build_doc_loader(assumptions);
106 let broken = docs.broken_links();
107 for b in broken {
108 eprintln!(
109 " broken link: {} -> {} (no page serves this slug)",
110 b.source_slug, b.target_slug
111 );
112 }
113 for c in docs.collisions() {
114 eprintln!(
115 " slug collision: {} ({} displaced by {})",
116 c.slug, c.displaced_section, c.winning_section
117 );
118 }
119 if broken.is_empty() {
120 println!(
121 "MNW_CHECK_DOCS: ok ({} collision(s) reported)",
122 docs.collisions().len()
123 );
124 std::process::exit(0);
125 }
126 println!("MNW_CHECK_DOCS: {} broken link(s)", broken.len());
127 std::process::exit(1);
128 }
129
130 let config = Config::from_env().expect("Failed to load configuration");
131 tracing::info!("Configuration loaded");
132
133 // Create database connection pool with health checks and lifecycle limits.
134 // - test_before_acquire: validates connections before use (catches stale/broken conns)
135 // - max_lifetime: rotates connections to prevent long-lived session issues
136 // - idle_timeout: prunes idle connections to free server resources
137 // - min_connections: keeps warm connections ready for immediate use
138 // - log_slow_statements: logs queries exceeding 100ms at WARN level
139 let connect_options: sqlx::postgres::PgConnectOptions =
140 config.database_url.parse().expect("Invalid DATABASE_URL");
141 let connect_options = connect_options
142 .log_statements(log::LevelFilter::Trace)
143 .log_slow_statements(log::LevelFilter::Warn, Duration::from_millis(100));
144
145 // Bound every query on every pooled connection: `statement_timeout` caps
146 // run time so a wedged query can't pin its connection forever (25 of those
147 // would exhaust the pool with no recovery, acquire_timeout only bounds
148 // getting a connection, not running one), and `lock_timeout` fails a
149 // lock-contended query fast instead of blocking indefinitely.
150 let statement_timeout_ms = constants::DB_STATEMENT_TIMEOUT_SECS * 1000;
151 let lock_timeout_ms = constants::DB_LOCK_TIMEOUT_SECS * 1000;
152 let db = PgPoolOptions::new()
153 .max_connections(constants::DB_POOL_MAX_CONNECTIONS)
154 .min_connections(constants::DB_POOL_MIN_CONNECTIONS)
155 .acquire_timeout(Duration::from_secs(constants::DB_ACQUIRE_TIMEOUT_SECS))
156 .max_lifetime(Duration::from_secs(constants::DB_MAX_LIFETIME_SECS))
157 .idle_timeout(Duration::from_secs(constants::DB_IDLE_TIMEOUT_SECS))
158 .test_before_acquire(true)
159 .after_connect(move |conn, _meta| {
160 Box::pin(async move {
161 use sqlx::Executor;
162 conn.execute(
163 format!(
164 "SET statement_timeout = {statement_timeout_ms}; SET lock_timeout = {lock_timeout_ms}"
165 )
166 .as_str(),
167 )
168 .await?;
169 Ok(())
170 })
171 })
172 .connect_with(connect_options)
173 .await
174 .expect("Failed to connect to database");
175
176 tracing::info!("Database connected");
177
178 // Run migrations, exit cleanly (code 2) on failure instead of panicking.
179 // This prevents systemd from crash-looping on migration errors (e.g. version
180 // conflicts after a bad deploy). WAM ticket alerts the operator.
181 if let Err(e) = sqlx::migrate!("./migrations").run(&db).await {
182 tracing::error!(error = %e, "Migration failed, exiting without restart");
183
184 // Best-effort WAM alert (DB is up, WAM may be reachable)
185 if let Ok(wam_url) = std::env::var("WAM_URL") {
186 let body = format!("Migration error on startup:\n{e}");
187 let mut req = reqwest::Client::new()
188 .post(format!("{wam_url}/tickets"))
189 .json(&serde_json::json!({
190 "title": "Migration failure, server not starting",
191 "body": body,
192 "priority": "critical",
193 "source": "migration-failure",
194 }))
195 .timeout(std::time::Duration::from_secs(5));
196 if let Ok(token) = std::env::var("WAM_TOKEN")
197 && !token.trim().is_empty()
198 {
199 req = req.bearer_auth(token.trim());
200 }
201 let _ = req.send().await;
202 }
203
204 // Exit code 2, systemd configured not to restart on this code
205 std::process::exit(2);
206 }
207
208 tracing::info!("Migrations complete");
209
210 // Example-marketplace seed for staging (testnot.work). Runs against the
211 // already-migrated DB, then exits, never during normal boot. Layered guards
212 // (opt-in env, host allowlist, no-real-users) make it refuse against prod.
213 // See `makenotwork::seed` and `_private/docs/mnw/testnot-example-seed.md`.
214 if std::env::args().any(|a| a == "--seed-examples") {
215 let opts = makenotwork::seed::SeedOptions::from_env(&config.host_url);
216
217 // Build the storage handles the media phase needs (main + public/CDN
218 // buckets). Both are `None` when S3 is unconfigured, the media phase then
219 // no-ops and items stay hidden. Done here, inside the one-shot seed block,
220 // so normal boot (which builds these later) is untouched.
221 async fn seed_client(
222 cfg: Option<&makenotwork::config::StorageConfig>,
223 host_url: &str,
224 ) -> Option<std::sync::Arc<dyn makenotwork::storage::StorageBackend>> {
225 let cfg = cfg?;
226 match S3Client::new(cfg, host_url).await {
227 Ok(client) => Some(std::sync::Arc::new(client) as _),
228 Err(e) => {
229 tracing::warn!(error = ?e, "seed: failed to init S3 storage; media skipped");
230 None
231 }
232 }
233 }
234 let media = makenotwork::seed::SeedMedia {
235 s3: seed_client(config.storage.as_ref(), &config.host_url).await,
236 public_s3: seed_client(config.public_storage.as_ref(), &config.host_url).await,
237 cdn_base_url: Some(config.cdn_base_url.clone()),
238 };
239
240 match makenotwork::seed::run(&db, &opts, &media).await {
241 Ok(()) => {
242 tracing::info!("example seed complete");
243 std::process::exit(0);
244 }
245 Err(e) => {
246 tracing::error!(error = %e, "example seed refused or failed");
247 std::process::exit(1);
248 }
249 }
250 }
251
252 // Create PostgreSQL-backed session store (persists across restarts)
253 let session_store = PostgresStore::new(db.clone());
254 session_store
255 .migrate()
256 .await
257 .expect("Failed to migrate session store");
258
259 // Continuously delete expired rows from the tower-sessions table. A session
260 // row is minted on every anonymous page render, so without this the table
261 // grows without bound under bot/crawler traffic and is read on every
262 // request. (The app's own `user_sessions` table is pruned by the daily
263 // scheduler; this covers the tower-sessions store, which the scheduler does
264 // not own.) Runs hourly on a background task.
265 {
266 use tower_sessions::ExpiredDeletion;
267 let deletion_store = session_store.clone();
268 tokio::task::spawn(async move {
269 if let Err(e) = deletion_store
270 .continuously_delete_expired(tokio::time::Duration::from_hours(1))
271 .await
272 {
273 tracing::error!(error = ?e, "tower-sessions expired-deletion task exited");
274 }
275 });
276 }
277
278 // In release mode, require HTTPS for session cookies (override with INSECURE_COOKIES=1 for staging)
279 let secure_cookies =
280 !cfg!(debug_assertions) && std::env::var("INSECURE_COOKIES").unwrap_or_default() != "1";
281 if secure_cookies {
282 tracing::info!("Session cookies configured for HTTPS (secure=true)");
283 } else if !cfg!(debug_assertions) {
284 tracing::warn!("Session cookies set to insecure (INSECURE_COOKIES=1)");
285 }
286
287 let session_layer = SessionManagerLayer::new(session_store)
288 .with_secure(secure_cookies)
289 .with_http_only(true)
290 .with_same_site(SameSite::Lax) // Lax allows session on top-level navigations (OAuth redirects)
291 .with_expiry(Expiry::OnInactivity(CookieDuration::days(
292 constants::SESSION_EXPIRY_DAYS,
293 )));
294
295 tracing::info!("PostgreSQL session store initialized");
296
297 // Initialize S3 client if storage is configured
298 let s3: Option<std::sync::Arc<dyn makenotwork::storage::StorageBackend>> = if let Some(
299 ref storage_config,
300 ) = config.storage
301 {
302 match S3Client::new(storage_config, &config.host_url).await {
303 Ok(client) => {
304 tracing::info!(bucket = %storage_config.bucket, "S3 storage initialized");
305 Some(std::sync::Arc::new(client))
306 }
307 Err(e) => {
308 tracing::warn!(error = ?e, "failed to initialize S3 storage, file uploads unavailable");
309 None
310 }
311 }
312 } else {
313 tracing::info!("S3 storage not configured. File uploads will be unavailable.");
314 None
315 };
316
317 // Initialize SyncKit blob S3 client if configured (separate bucket)
318 let synckit_s3: Option<std::sync::Arc<dyn makenotwork::storage::StorageBackend>> = if let Some(
319 ref synckit_storage_config,
320 ) =
321 config.synckit_storage
322 {
323 match S3Client::new(synckit_storage_config, &config.host_url).await {
324 Ok(client) => {
325 tracing::info!(bucket = %synckit_storage_config.bucket, "SyncKit blob storage initialized");
326 Some(std::sync::Arc::new(client))
327 }
328 Err(e) => {
329 tracing::warn!(error = ?e, "Failed to initialize SyncKit blob storage");
330 None
331 }
332 }
333 } else {
334 tracing::info!("SyncKit blob storage not configured");
335 None
336 };
337
338 // Initialize public (CDN-served) bucket client if configured. Holds only
339 // promoted image content; the scan worker copies Clean covers/gallery here.
340 let public_s3: Option<std::sync::Arc<dyn makenotwork::storage::StorageBackend>> = if let Some(
341 ref public_storage_config,
342 ) =
343 config.public_storage
344 {
345 match S3Client::new(public_storage_config, &config.host_url).await {
346 Ok(client) => {
347 tracing::info!(bucket = %public_storage_config.bucket, "Public S3 bucket initialized");
348 Some(std::sync::Arc::new(client))
349 }
350 Err(e) => {
351 tracing::warn!(error = ?e, "Failed to initialize public S3 bucket");
352 None
353 }
354 }
355 } else {
356 tracing::info!("Public S3 bucket not configured");
357 None
358 };
359
360 // Initialize Stripe client if configured
361 let stripe: Option<std::sync::Arc<dyn makenotwork::payments::PaymentProvider>> = if let Some(
362 ref stripe_config,
363 ) =
364 config.stripe
365 {
366 match StripeClient::new(stripe_config) {
367 Ok(client) => {
368 tracing::info!("Stripe payments initialized");
369 Some(std::sync::Arc::new(client)
370 as std::sync::Arc<dyn makenotwork::payments::PaymentProvider>)
371 }
372 Err(e) => {
373 // Invalid client config is a boot invariant violation, exit
374 // without restart rather than run with payments silently broken.
375 tracing::error!(error = %e, "Failed to build Stripe client, exiting without restart");
376 std::process::exit(2);
377 }
378 }
379 } else {
380 tracing::info!("Stripe not configured. Payments will be unavailable.");
381 None
382 };
383
384 // Initialize email client (logs in dev mode when POSTMARK_TOKEN is not set)
385 let email = EmailClient::new(EmailConfig::from_env(), Some(db.clone()));
386
387 // Load business assumptions (single source of truth for figures in the docs).
388 // Validation failure aborts startup, we don't want to serve stale numbers.
389 let assumptions = load_assumptions().unwrap_or_else(|e| panic!("{e}"));
390
391 // Load documentation pages from disk. The same builder backs the DB-free
392 // MNW_CHECK_DOCS integrity check above, so what boots and what is checked
393 // are the same corpus under the same config.
394 let docs = std::sync::Arc::new(build_doc_loader(assumptions.clone()));
395
396 // Initialize file scanning pipeline (optional). If scanning is configured,
397 // assert at least one AV layer is actually live, otherwise the FailOpen
398 // policy on ClamAV silently passes everything as Clean and the deploy
399 // looks healthy while shipping zero real coverage. Refuse to boot.
400 let scanner = if let Some(ref scan_config) = config.scan {
401 match ScanPipeline::new(scan_config) {
402 Ok(s) => match s.assert_live().await {
403 Ok(()) => {
404 tracing::info!("File scanning enabled");
405 Some(std::sync::Arc::new(s))
406 }
407 Err(e) => panic!("Scanning configured but no live AV layer: {e}"),
408 },
409 Err(e) => {
410 tracing::warn!(error = %e, "File scanning disabled");
411 None
412 }
413 }
414 } else {
415 tracing::info!("File scanning not configured");
416 None
417 };
418
419 // Initialize WebAuthn from HOST_URL (passkey / passwordless login)
420 let rp_origin = url::Url::parse(&config.host_url).expect("HOST_URL is not a valid URL");
421 let rp_id = rp_origin
422 .host_str()
423 .expect("HOST_URL has no host")
424 .to_string();
425 let webauthn = std::sync::Arc::new(
426 WebauthnBuilder::new(&rp_id, &rp_origin)
427 .expect("Failed to create WebauthnBuilder")
428 .rp_name("Makenotwork")
429 .build()
430 .expect("Failed to build Webauthn"),
431 );
432 tracing::info!("WebAuthn initialized (rp_id={rp_id})");
433
434 // Initialize syntax highlighter if git repos path is configured
435 let syntax = if config.build.git_repos_path.is_some() {
436 tracing::info!(path = ?config.build.git_repos_path, "Git source browser enabled");
437 Some(std::sync::Arc::new(
438 makenotwork::git::SyntaxHighlighter::new(),
439 ))
440 } else {
441 tracing::info!("Git source browser not configured (GIT_REPOS_PATH unset)");
442 None
443 };
444
445 // Construct MT client when both mt_base_url and internal_shared_secret are set
446 let mt_client = match (
447 &config.integrations.mt_base_url,
448 &config.integrations.internal_shared_secret,
449 ) {
450 (Some(base_url), Some(secret)) => {
451 tracing::info!(base_url = %base_url, "MT integration enabled");
452 Some(makenotwork::mt_client::MtClient::new(
453 base_url.clone(),
454 secret.clone(),
455 ))
456 }
457 _ => {
458 tracing::info!(
459 "MT integration not configured (need MT_BASE_URL + INTERNAL_SHARED_SECRET)"
460 );
461 None
462 }
463 };
464
465 // WAM ticket manager client (tailnet-only, for operational alerts). The
466 // optional WAM_TOKEN is the shared secret WAM enforces when set.
467 let wam = config.integrations.wam_url.as_ref().map(|url| {
468 tracing::info!(url = %url, "WAM integration enabled");
469 makenotwork::wam_client::WamClient::new(url.clone(), std::env::var("WAM_TOKEN").ok())
470 });
471
472 // Warm custom domain cache
473 let domain_cache = std::sync::Arc::new(dashmap::DashMap::new());
474 match makenotwork::db::custom_domains::get_all_verified_domains(&db).await {
475 Ok(domains) => {
476 for d in &domains {
477 domain_cache.insert(d.domain.clone(), d.user_id);
478 }
479 if !domains.is_empty() {
480 tracing::info!(count = domains.len(), "Custom domain cache warmed");
481 }
482 }
483 Err(e) => {
484 tracing::warn!(error = ?e, "Failed to warm custom domain cache");
485 }
486 }
487
488 // Shutdown broadcast: dropped at graceful-shutdown time to signal the
489 // background pool, monitor, scheduler, and scan workers to drain and stop.
490 let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(());
491 let page_view_tx = makenotwork::db::page_views::spawn_batcher(db.clone());
492 let (bg, bg_handle) = makenotwork::background::spawn_pool(shutdown_tx.subscribe());
493
494 // Derive pricing from the loaded assumptions. Installing the process-global
495 // TierPrices is an explicit, ordered step here (not buried in AppState
496 // construction): CreatorTier accessors (price_cents, max_file_bytes,
497 // max_storage_bytes) read the global, so it must be installed before any code
498 // path constructs a CreatorTier and calls one of them.
499 let tier_prices = makenotwork::tier_prices::TierPrices::from_assumptions(&assumptions);
500 tier_prices.clone().install_global();
501 let runway_config = makenotwork::tier_prices::RunwayConfig::from_assumptions(&assumptions);
502 let pricing_comparison =
503 makenotwork::pricing_comparison::PricingComparison::load(assumptions_path());
504
505 let state = AppState::build(makenotwork::AppStateParts {
506 db,
507 config: config.clone(),
508 storage: AppStorage {
509 s3,
510 synckit_s3,
511 public_s3,
512 },
513 stripe,
514 email,
515 docs,
516 tier_prices,
517 runway_config,
518 pricing_comparison,
519 scanner,
520 webauthn,
521 syntax,
522 mt_client,
523 wam,
524 domain_cache,
525 metrics_handle: Some(makenotwork::metrics::init()),
526 page_view_tx,
527 bg,
528 });
529
530 // Log active features at startup
531 tracing::info!(
532 s3 = state.storage.s3.is_some(),
533 synckit_s3 = state.storage.synckit_s3.is_some(),
534 stripe = state.stripe.is_some(),
535 scanner = state.scanner.is_some(),
536 mt = state.mt_client.is_some(),
537 wam = state.wam.is_some(),
538 git = state.config.build.git_repos_path.is_some(),
539 "Active features"
540 );
541
542 // Start background health monitor and scheduler
543 let _monitor_handle =
544 makenotwork::monitor::spawn_monitor(axum::extract::FromRef::from_ref(&state), shutdown_rx);
545 let scheduler_shutdown_rx = shutdown_tx.subscribe();
546 let scheduler_handle =
547 makenotwork::scheduler::spawn_scheduler(state.clone(), scheduler_shutdown_rx);
548
549 // Per-instance (not scheduler-gated, so every instance in a rolling deploy
550 // agrees): decides whether the footer links to /changelog at all.
551 let _changelog_handle =
552 makenotwork::changelog::spawn_refresher(state.db.clone(), shutdown_tx.subscribe());
553
554 // Start scan worker pool. Only meaningful if a scanner is configured;
555 // otherwise enqueue_scan_for never enqueues (trust-gate fast path).
556 if let (Some(scanner), Some(s3_for_workers)) = (state.scanner.clone(), state.storage.s3.clone())
557 {
558 let scan_ctx = std::sync::Arc::new(makenotwork::scanning::worker::WorkerContext {
559 db: state.db.clone(),
560 s3: s3_for_workers,
561 pipeline: scanner,
562 scan_semaphore: state.limiters.scan_semaphore.clone(),
563 wam: state.wam.clone(),
564 bg: state.bg.clone(),
565 cloudflare: makenotwork::cloudflare::CloudflarePurger::from_env(),
566 cdn_base_url: std::sync::Arc::from(state.config.cdn_base_url.as_str()),
567 synckit_s3: state.storage.synckit_s3.clone(),
568 public_s3: state.storage.public_s3.clone(),
569 });
570 let worker_count = makenotwork::constants::SCAN_WORKER_COUNT;
571 let worker_shutdown_rx = shutdown_tx.subscribe();
572 makenotwork::scanning::worker::spawn_pool(worker_count, &scan_ctx, worker_shutdown_rx);
573 tracing::info!(worker_count, "scan worker pool started");
574
575 let report = makenotwork::scanning::spool::reap_all(std::path::Path::new(
576 makenotwork::constants::SCAN_SPOOL_DIR,
577 ));
578 if report.deleted > 0 || report.errors > 0 {
579 tracing::info!(
580 deleted = report.deleted,
581 errors = report.errors,
582 "scan spool startup reaper completed"
583 );
584 }
585 }
586
587 // Build router (shared with integration tests via lib.rs)
588 let app = build_app(&state, session_layer)
589 // Request ID: propagate → trace → set (Axum applies inside-out)
590 .layer(PropagateRequestIdLayer::x_request_id())
591 .layer(
592 TraceLayer::new_for_http()
593 .make_span_with(|request: &Request<_>| {
594 let request_id = request
595 .headers()
596 .get("x-request-id")
597 .and_then(|v| v.to_str().ok())
598 .unwrap_or("-");
599
600 tracing::info_span!(
601 "request",
602 method = %request.method(),
603 uri = %request.uri(),
604 request_id = %request_id,
605 user_id = tracing::field::Empty,
606 )
607 })
608 .on_response(
609 |response: &axum::http::Response<_>,
610 latency: Duration,
611 _span: &tracing::Span| {
612 let status = response.status().as_u16();
613 if status >= 500 {
614 tracing::error!(
615 status,
616 latency_ms = latency.as_millis() as u64,
617 "response"
618 );
619 } else if status >= 400 {
620 tracing::warn!(
621 status,
622 latency_ms = latency.as_millis() as u64,
623 "response"
624 );
625 } else {
626 tracing::debug!(
627 status,
628 latency_ms = latency.as_millis() as u64,
629 "response"
630 );
631 }
632 },
633 ),
634 )
635 .layer(SetRequestIdLayer::x_request_id(MakeRequestUuid));
636
637 let addr = config.socket_addr();
638 tracing::info!(addr = %addr, "listening");
639
640 let listener = tokio::net::TcpListener::bind(addr)
641 .await
642 .expect("Failed to bind to address");
643
644 // Use into_make_service_with_connect_info to provide peer IP for rate limiting.
645 // After the shutdown signal fires, in-flight requests drain, then the
646 // scheduler + background pool drain concurrently; the hard-exit timer
647 // (SHUTDOWN_HARD_EXIT_SECS, armed at signal receipt) is the backstop that
648 // frees a hung restart. The budget is reconciled so the pool drain fits
649 // before the hard exit (see the shutdown constants below).
650 // Bound the in-flight request drain. `with_graceful_shutdown` alone waits for
651 // in-flight requests with NO deadline, so a single hung request stalls
652 // shutdown until the hard-exit kills the process, taking the post-serve pool
653 // drain with it (audit Run 24 A4). Fire a one-shot the instant the signal
654 // arrives, then give the drain at most SHUTDOWN_REQUEST_DRAIN_SECS from that
655 // point before abandoning it and proceeding to the pool drain. (The timer must
656 // start at signal time, not server-start time, hence the one-shot.)
657 let (signal_tx, signal_rx) = tokio::sync::oneshot::channel::<()>();
658 let signal_tx = std::sync::Mutex::new(Some(signal_tx));
659 let serve = axum::serve(
660 listener,
661 app.into_make_service_with_connect_info::<std::net::SocketAddr>(),
662 )
663 .with_graceful_shutdown(async move {
664 shutdown_signal().await;
665 if let Some(tx) = signal_tx.lock().unwrap().take() {
666 let _ = tx.send(());
667 }
668 })
669 .into_future();
670 tokio::pin!(serve);
671
672 let drain_deadline = async {
673 let _ = signal_rx.await; // resolves only once the signal fires
674 tokio::time::sleep(Duration::from_secs(SHUTDOWN_REQUEST_DRAIN_SECS)).await;
675 };
676 tokio::select! {
677 r = &mut serve => r.expect("Server error"),
678 () = drain_deadline => tracing::warn!(
679 "in-flight request drain exceeded {SHUTDOWN_REQUEST_DRAIN_SECS}s; \
680 abandoning it and proceeding to the pool drain"
681 ),
682 }
683
684 drop(shutdown_tx); // signal monitor, scheduler, and scan workers to stop
685
686 // Drain the scheduler's in-flight tick (platform-credit settlement,
687 // pending-S3-deletion retries, webhook re-drives) and the fire-and-forget
688 // background pool (queued emails / cache purges / MT thread creations)
689 // together. Both are engineered idempotent, so a truncated drain is
690 // recoverable, but awaiting a bounded window lets in-progress work finish
691 // cleanly. They run CONCURRENTLY under one shared budget so the pool drain
692 // isn't starved by a slow scheduler tick before the hard-exit fires: the old
693 // sequential ≤5s + ≤5s could reach 10s and let the hard-exit truncate the
694 // second drain (audit Run 23 F4). If the shared window overruns we proceed
695 // and rely on idempotency as the backstop.
696 let drains = async { tokio::join!(scheduler_handle, bg_handle) };
697 match tokio::time::timeout(Duration::from_secs(SHUTDOWN_POOL_DRAIN_SECS), drains).await {
698 Ok((scheduler_res, bg_res)) => {
699 match scheduler_res {
700 Ok(()) => tracing::info!("scheduler drained"),
701 Err(e) => tracing::warn!(error = ?e, "scheduler task ended abnormally on shutdown"),
702 }
703 match bg_res {
704 Ok(()) => tracing::info!("background pool drained"),
705 Err(e) => {
706 tracing::warn!(error = ?e, "background pool task ended abnormally on shutdown");
707 }
708 }
709 }
710 Err(_) => tracing::warn!(
711 "scheduler + background pool did not drain within {SHUTDOWN_POOL_DRAIN_SECS}s of \
712 shutdown; proceeding (work is idempotent)"
713 ),
714 }
715
716 tracing::info!("Server shut down gracefully");
717 }
718
719 /// Filesystem path to the business-assumptions TOML, `ASSUMPTIONS_PATH` or its
720 /// default. Single source for the default so every reader agrees.
721 fn assumptions_path() -> String {
722 std::env::var("ASSUMPTIONS_PATH")
723 .unwrap_or_else(|_| "docs/business/assumptions.toml".to_string())
724 }
725
726 /// Load and validate business assumptions. `Err` carries a human-readable
727 /// message; normal startup turns it into a panic, the check modes into a
728 /// sentinel line. Shared so the two paths validate identically.
729 fn load_assumptions() -> Result<std::sync::Arc<Assumptions>, String> {
730 let assumptions_path = assumptions_path();
731 let a = Assumptions::load(&assumptions_path)
732 .map_err(|e| format!("failed to load assumptions from {assumptions_path}: {e}"))?;
733 a.validate()
734 .map_err(|e| format!("assumptions validation failed:\n{e}"))?;
735 tracing::info!(path = %assumptions_path, "assumptions loaded and validated");
736 Ok(std::sync::Arc::new(a))
737 }
738
739 /// Build the documentation loader from disk with the production config.
740 ///
741 /// Shared by normal startup and the `MNW_CHECK_DOCS` integrity check so both
742 /// see the exact same sections, link prefix, examples path, and assumption
743 /// substitution, so the broken-link report can't drift from what is served.
744 fn build_doc_loader(assumptions: std::sync::Arc<Assumptions>) -> DocLoader {
745 let docs_path = std::env::var("DOCS_PATH").unwrap_or_else(|_| "site-docs/public".to_string());
746 DocLoader::load(
747 std::path::Path::new(&docs_path),
748 &DocLoaderConfig {
749 sections: vec![
750 ("about".to_string(), "About".to_string()),
751 ("guide".to_string(), "Guide".to_string()),
752 ("developer".to_string(), "Developer".to_string()),
753 ("legal".to_string(), "Legal".to_string()),
754 ("support".to_string(), "Support".to_string()),
755 ("tech".to_string(), "Tech".to_string()),
756 ],
757 link_prefix: "/docs".to_string(),
758 unpublished_pattern: Some("unpublished/".to_string()),
759 examples_path: Some(std::path::Path::new(&docs_path).join("../examples")),
760 pre_process: Some(Box::new(move |md: &str| {
761 assumptions.substitute(md).map_err(|e| e.to_string())
762 })),
763 },
764 )
765 }
766
767 /// Shutdown budget, reconciled so the background-pool drain can't be truncated
768 /// by the hard-exit deadline (audit Run 23 F4). The hard-exit timer arms at
769 /// signal receipt and must outlast the in-flight request drain PLUS the
770 /// post-serve drain window, otherwise the later drain is cut off mid-flight.
771 /// The two post-serve drains run *concurrently* (see `join!` below), so they
772 /// share the pool-drain window rather than summing to twice it.
773 const SHUTDOWN_REQUEST_DRAIN_SECS: u64 = 10;
774 const SHUTDOWN_POOL_DRAIN_SECS: u64 = 5;
775 /// Hard-exit backstop, defined as its own ceiling (not the sum of the two
776 /// drains) with margin above them, the process is force-killed only if BOTH
777 /// bounded drains overrun their budgets. The margin also makes the invariant
778 /// below a real check rather than the tautology it was when this equalled the
779 /// sum (audit Run 24 A4).
780 const SHUTDOWN_HARD_EXIT_SECS: u64 = 20;
781 const _: () = assert!(
782 SHUTDOWN_HARD_EXIT_SECS > SHUTDOWN_REQUEST_DRAIN_SECS + SHUTDOWN_POOL_DRAIN_SECS,
783 "hard-exit must outlast the bounded request + pool drains, with margin"
784 );
785
786 /// Wait for SIGINT (Ctrl-C) or SIGTERM, then return to trigger graceful shutdown.
787 /// In-flight requests get up to `SHUTDOWN_REQUEST_DRAIN_SECS` to complete, then
788 /// the post-serve drains get `SHUTDOWN_POOL_DRAIN_SECS`; the hard-exit timer
789 /// (`SHUTDOWN_HARD_EXIT_SECS`) is the backstop that frees a hung restart.
790 async fn shutdown_signal() {
791 use tokio::signal;
792
793 let ctrl_c = async {
794 signal::ctrl_c()
795 .await
796 .expect("Failed to install Ctrl+C handler");
797 };
798
799 #[cfg(unix)]
800 let terminate = async {
801 signal::unix::signal(signal::unix::SignalKind::terminate())
802 .expect("Failed to install SIGTERM handler")
803 .recv()
804 .await;
805 };
806
807 #[cfg(not(unix))]
808 let terminate = std::future::pending::<()>();
809
810 tokio::select! {
811 () = ctrl_c => tracing::info!("Received SIGINT, starting graceful shutdown"),
812 () = terminate => tracing::info!("Received SIGTERM, starting graceful shutdown"),
813 }
814
815 // Spawn the hard-exit backstop: if the full graceful drain (request drain +
816 // concurrent pool drain) overruns its reconciled budget, force exit so a hung
817 // connection can't block a deployment restart indefinitely.
818 tokio::spawn(async {
819 tokio::time::sleep(Duration::from_secs(SHUTDOWN_HARD_EXIT_SECS)).await;
820 eprintln!("Graceful shutdown timed out after {SHUTDOWN_HARD_EXIT_SECS}s, forcing exit");
821 std::process::exit(1);
822 });
823 }
824