Skip to main content

max / makenotwork

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