Skip to main content

max / makenotwork

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