Skip to main content

max / makenotwork

41.0 KB · 974 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. Stripe implements every
418 // capability extension, so one `Arc` becomes both the base provider and
419 // the whole capability set.
420 let (stripe, payment_caps): (
421 Option<std::sync::Arc<dyn makenotwork::payments::PaymentProvider>>,
422 makenotwork::payments::PaymentCapabilities,
423 ) = if let Some(ref stripe_config) = config.stripe {
424 match StripeClient::new(stripe_config) {
425 Ok(client) => {
426 tracing::info!("Stripe payments initialized");
427 let client = std::sync::Arc::new(client);
428 (
429 Some(client.clone()
430 as std::sync::Arc<dyn makenotwork::payments::PaymentProvider>),
431 makenotwork::payments::PaymentCapabilities::all(client),
432 )
433 }
434 Err(e) => {
435 // Invalid client config is a boot invariant violation, exit
436 // without restart rather than run with payments silently broken.
437 tracing::error!(error = %e, "Failed to build Stripe client, exiting without restart");
438 std::process::exit(2);
439 }
440 }
441 } else {
442 tracing::info!("Stripe not configured. Payments will be unavailable.");
443 (None, makenotwork::payments::PaymentCapabilities::default())
444 };
445
446 // Initialize email client (logs in dev mode when POSTMARK_TOKEN is not set)
447 let email = EmailClient::new(EmailConfig::from_env(), Some(db.clone()));
448
449 // Load business assumptions (single source of truth for figures in the docs).
450 // Validation failure aborts startup, we don't want to serve stale numbers.
451 let assumptions = makenotwork::site_docs::load_assumptions().unwrap_or_else(|e| panic!("{e}"));
452
453 // Load documentation pages from disk. The same builder backs the DB-free
454 // MNW_CHECK_DOCS integrity check above, so what boots and what is checked
455 // are the same corpus under the same config.
456 let docs = std::sync::Arc::new(makenotwork::site_docs::build_doc_loader(
457 assumptions.clone(),
458 ));
459
460 // Initialize file scanning pipeline (optional). If scanning is configured,
461 // assert at least one AV layer is actually live, otherwise the FailOpen
462 // policy on ClamAV silently passes everything as Clean and the deploy
463 // looks healthy while shipping zero real coverage. Refuse to boot.
464 let scanner = if let Some(ref scan_config) = config.scan {
465 match ScanPipeline::new(scan_config) {
466 Ok(s) => match s.assert_live().await {
467 Ok(()) => {
468 tracing::info!("File scanning enabled");
469 Some(std::sync::Arc::new(s))
470 }
471 Err(e) => panic!("Scanning configured but no live AV layer: {e}"),
472 },
473 Err(e) => {
474 tracing::warn!(error = %e, "File scanning disabled");
475 None
476 }
477 }
478 } else {
479 tracing::info!("File scanning not configured");
480 None
481 };
482
483 // Initialize WebAuthn from HOST_URL (passkey / passwordless login)
484 let rp_origin = url::Url::parse(&config.host_url).expect("HOST_URL is not a valid URL");
485 let rp_id = rp_origin
486 .host_str()
487 .expect("HOST_URL has no host")
488 .to_string();
489 let webauthn = std::sync::Arc::new(
490 WebauthnBuilder::new(&rp_id, &rp_origin)
491 .expect("Failed to create WebauthnBuilder")
492 .rp_name("Makenotwork")
493 .build()
494 .expect("Failed to build Webauthn"),
495 );
496 tracing::info!("WebAuthn initialized (rp_id={rp_id})");
497
498 // Initialize syntax highlighter if git repos path is configured
499 let syntax = if config.build.git_repos_path.is_some() {
500 tracing::info!(path = ?config.build.git_repos_path, "Git source browser enabled");
501 Some(std::sync::Arc::new(
502 makenotwork::git::SyntaxHighlighter::new(),
503 ))
504 } else {
505 tracing::info!("Git source browser not configured (GIT_REPOS_PATH unset)");
506 None
507 };
508
509 // Construct MT client when both mt_base_url and internal_shared_secret are set
510 let mt_client = match (
511 &config.integrations.mt_base_url,
512 &config.integrations.internal_shared_secret,
513 ) {
514 (Some(base_url), Some(secret)) => {
515 tracing::info!(base_url = %base_url, "MT integration enabled");
516 Some(makenotwork::mt_client::MtClient::new(
517 base_url.clone(),
518 secret.clone(),
519 ))
520 }
521 _ => {
522 tracing::info!(
523 "MT integration not configured (need MT_BASE_URL + INTERNAL_SHARED_SECRET)"
524 );
525 None
526 }
527 };
528
529 // WAM ticket manager client (tailnet-only, for operational alerts). The
530 // optional WAM_TOKEN is the shared secret WAM enforces when set.
531 let wam = config.integrations.wam_url.as_ref().map(|url| {
532 tracing::info!(url = %url, "WAM integration enabled");
533 makenotwork::wam_client::WamClient::new(url.clone(), std::env::var("WAM_TOKEN").ok())
534 });
535
536 // Warm custom domain cache
537 let domain_cache = std::sync::Arc::new(dashmap::DashMap::new());
538 match makenotwork::db::custom_domains::get_all_verified_domains(&db).await {
539 Ok(domains) => {
540 for d in &domains {
541 domain_cache.insert(d.domain.clone(), d.user_id);
542 }
543 if !domains.is_empty() {
544 tracing::info!(count = domains.len(), "Custom domain cache warmed");
545 }
546 }
547 Err(e) => {
548 tracing::warn!(error = ?e, "Failed to warm custom domain cache");
549 }
550 }
551
552 // Shutdown broadcast: dropped at graceful-shutdown time to signal the
553 // background pool, monitor, scheduler, and scan workers to drain and stop.
554 let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(());
555 let page_view_tx = makenotwork::db::page_views::spawn_batcher(db.clone());
556 let (bg, bg_handle) = makenotwork::background::spawn_pool(shutdown_tx.subscribe());
557
558 // Derive pricing from the loaded assumptions. Installing the process-global
559 // TierPrices is an explicit, ordered step here (not buried in AppState
560 // construction): CreatorTier accessors (price_cents, max_file_bytes,
561 // max_storage_bytes) read the global, so it must be installed before any code
562 // path constructs a CreatorTier and calls one of them.
563 let tier_prices = makenotwork::tier_prices::TierPrices::from_assumptions(&assumptions);
564 tier_prices.clone().install_global();
565 let runway_config = makenotwork::tier_prices::RunwayConfig::from_assumptions(&assumptions);
566 let fee_calculator = makenotwork::fee_calculator::FeeCalculator::load(
567 makenotwork::site_docs::assumptions_path(),
568 );
569
570 let state = AppState::build(makenotwork::AppStateParts {
571 db,
572 config: config.clone(),
573 storage: AppStorage {
574 s3,
575 synckit_s3,
576 public_s3,
577 rpm_s3,
578 },
579 payments: stripe,
580 payment_caps,
581 email,
582 docs,
583 tier_prices,
584 runway_config,
585 fee_calculator,
586 scanner,
587 webauthn,
588 syntax,
589 mt_client,
590 wam,
591 domain_cache,
592 metrics_handle: Some(makenotwork::metrics::init()),
593 page_view_tx,
594 bg,
595 });
596
597 // Log active features at startup
598 tracing::info!(
599 s3 = state.storage.s3.is_some(),
600 synckit_s3 = state.storage.synckit_s3.is_some(),
601 stripe = state.payments.is_some(),
602 scanner = state.scanner.is_some(),
603 mt = state.mt_client.is_some(),
604 wam = state.wam.is_some(),
605 git = state.config.build.git_repos_path.is_some(),
606 "Active features"
607 );
608
609 // Start background health monitor and scheduler
610 let _monitor_handle =
611 makenotwork::monitor::spawn_monitor(axum::extract::FromRef::from_ref(&state), shutdown_rx);
612 let scheduler_shutdown_rx = shutdown_tx.subscribe();
613 let scheduler_handle =
614 makenotwork::scheduler::spawn_scheduler(state.clone(), scheduler_shutdown_rx);
615
616 // Per-instance (not scheduler-gated, so every instance in a rolling deploy
617 // agrees): decides whether the footer links to /changelog at all.
618 let _changelog_handle =
619 makenotwork::changelog::spawn_refresher(state.db.clone(), shutdown_tx.subscribe());
620
621 // Start scan worker pool. Only meaningful if a scanner is configured;
622 // otherwise enqueue_scan_for never enqueues (trust-gate fast path).
623 if let (Some(scanner), Some(s3_for_workers)) = (state.scanner.clone(), state.storage.s3.clone())
624 {
625 let scan_ctx = std::sync::Arc::new(makenotwork::scanning::worker::WorkerContext {
626 db: state.db.clone(),
627 s3: s3_for_workers,
628 pipeline: scanner,
629 scan_semaphore: state.limiters.scan_semaphore.clone(),
630 wam: state.wam.clone(),
631 bg: state.bg.clone(),
632 cloudflare: makenotwork::cloudflare::CloudflarePurger::from_env(),
633 cdn_base_url: std::sync::Arc::from(state.config.cdn_base_url.as_str()),
634 synckit_s3: state.storage.synckit_s3.clone(),
635 public_s3: state.storage.public_s3.clone(),
636 config: state.config.clone(),
637 });
638 let worker_count = makenotwork::constants::SCAN_WORKER_COUNT;
639 let worker_shutdown_rx = shutdown_tx.subscribe();
640 makenotwork::scanning::worker::spawn_pool(worker_count, &scan_ctx, worker_shutdown_rx);
641 tracing::info!(worker_count, "scan worker pool started");
642
643 let report = makenotwork::scanning::spool::reap_all(std::path::Path::new(
644 makenotwork::constants::SCAN_SPOOL_DIR,
645 ));
646 if report.deleted > 0 || report.errors > 0 {
647 tracing::info!(
648 deleted = report.deleted,
649 errors = report.errors,
650 "scan spool startup reaper completed"
651 );
652 }
653 }
654
655 // Security signals need somewhere to send an alert. Before the router, so
656 // nothing the server answers is counted against an uninstalled sink.
657 makenotwork::security_signals::install(state.db.clone(), state.email.clone());
658
659 // Build router (shared with integration tests via lib.rs)
660 let app = build_app(&state, session_layer)
661 // Outside every per-route limiter, so a 429 the governor returned is
662 // seen here. Counting only; the alert, if any, is spawned off the
663 // request path.
664 .layer(axum::middleware::from_fn(
665 |req: Request<axum::body::Body>, next: axum::middleware::Next| async move {
666 // Read the header rather than calling `extract_client_ip`: that
667 // helper counts absences to warn about a missing Cloudflare
668 // proxy, and a per-response call would double every count.
669 let ip = req
670 .headers()
671 .get("cf-connecting-ip")
672 .and_then(|v| v.to_str().ok())
673 .and_then(|s| s.split(',').next())
674 .map(|s| s.trim().to_string())
675 .filter(|s| !s.is_empty());
676 let response = next.run(req).await;
677 makenotwork::security_signals::note_response(
678 response.status().as_u16(),
679 ip.as_deref(),
680 );
681 response
682 },
683 ))
684 // Request ID: propagate → trace → set (Axum applies inside-out)
685 .layer(PropagateRequestIdLayer::x_request_id())
686 .layer(
687 TraceLayer::new_for_http()
688 .make_span_with(|request: &Request<_>| {
689 let request_id = request
690 .headers()
691 .get("x-request-id")
692 .and_then(|v| v.to_str().ok())
693 .unwrap_or("-");
694
695 tracing::info_span!(
696 "request",
697 method = %request.method(),
698 uri = %request.uri(),
699 request_id = %request_id,
700 user_id = tracing::field::Empty,
701 )
702 })
703 .on_response(
704 |response: &axum::http::Response<_>,
705 latency: Duration,
706 _span: &tracing::Span| {
707 let status = response.status().as_u16();
708 if status >= 500 {
709 tracing::error!(
710 status,
711 latency_ms = latency.as_millis() as u64,
712 "response"
713 );
714 } else if status >= 400 {
715 tracing::warn!(
716 status,
717 latency_ms = latency.as_millis() as u64,
718 "response"
719 );
720 } else {
721 tracing::debug!(
722 status,
723 latency_ms = latency.as_millis() as u64,
724 "response"
725 );
726 }
727 },
728 ),
729 )
730 .layer(SetRequestIdLayer::x_request_id(MakeRequestUuid));
731
732 let addr = config.socket_addr();
733 tracing::info!(addr = %addr, "listening");
734
735 let listener = tokio::net::TcpListener::bind(addr)
736 .await
737 .expect("Failed to bind to address");
738
739 // Use into_make_service_with_connect_info to provide peer IP for rate limiting.
740 // After the shutdown signal fires, in-flight requests drain, then the
741 // scheduler + background pool drain concurrently; the hard-exit timer
742 // (SHUTDOWN_HARD_EXIT_SECS, armed at signal receipt) is the backstop that
743 // frees a hung restart. The budget is reconciled so the pool drain fits
744 // before the hard exit (see the shutdown constants below).
745 // Bound the in-flight request drain. `with_graceful_shutdown` alone waits for
746 // in-flight requests with NO deadline, so a single hung request stalls
747 // shutdown until the hard-exit kills the process, taking the post-serve pool
748 // drain with it (audit Run 24 A4). Fire a one-shot the instant the signal
749 // arrives, then give the drain at most SHUTDOWN_REQUEST_DRAIN_SECS from that
750 // point before abandoning it and proceeding to the pool drain. (The timer must
751 // start at signal time, not server-start time, hence the one-shot.)
752 let (signal_tx, signal_rx) = tokio::sync::oneshot::channel::<()>();
753 let signal_tx = std::sync::Mutex::new(Some(signal_tx));
754 let serve = axum::serve(
755 listener,
756 app.into_make_service_with_connect_info::<std::net::SocketAddr>(),
757 )
758 .with_graceful_shutdown(async move {
759 shutdown_signal().await;
760 if let Some(tx) = signal_tx.lock().unwrap().take() {
761 let _ = tx.send(());
762 }
763 })
764 .into_future();
765 tokio::pin!(serve);
766
767 let drain_deadline = async {
768 let _ = signal_rx.await; // resolves only once the signal fires
769 tokio::time::sleep(Duration::from_secs(SHUTDOWN_REQUEST_DRAIN_SECS)).await;
770 };
771 tokio::select! {
772 r = &mut serve => r.expect("Server error"),
773 () = drain_deadline => tracing::warn!(
774 "in-flight request drain exceeded {SHUTDOWN_REQUEST_DRAIN_SECS}s; \
775 abandoning it and proceeding to the pool drain"
776 ),
777 }
778
779 drop(shutdown_tx); // signal monitor, scheduler, and scan workers to stop
780
781 // Drain the scheduler's in-flight tick (platform-credit settlement,
782 // pending-S3-deletion retries, webhook re-drives) and the fire-and-forget
783 // background pool (queued emails / cache purges / MT thread creations)
784 // together. Both are engineered idempotent, so a truncated drain is
785 // recoverable, but awaiting a bounded window lets in-progress work finish
786 // cleanly. They run CONCURRENTLY under one shared budget so the pool drain
787 // isn't starved by a slow scheduler tick before the hard-exit fires: the old
788 // sequential ≤5s + ≤5s could reach 10s and let the hard-exit truncate the
789 // second drain (audit Run 23 F4). If the shared window overruns we proceed
790 // and rely on idempotency as the backstop.
791 let drains = async { tokio::join!(scheduler_handle, bg_handle) };
792 match tokio::time::timeout(Duration::from_secs(SHUTDOWN_POOL_DRAIN_SECS), drains).await {
793 Ok((scheduler_res, bg_res)) => {
794 match scheduler_res {
795 Ok(()) => tracing::info!("scheduler drained"),
796 Err(e) => tracing::warn!(error = ?e, "scheduler task ended abnormally on shutdown"),
797 }
798 match bg_res {
799 Ok(()) => tracing::info!("background pool drained"),
800 Err(e) => {
801 tracing::warn!(error = ?e, "background pool task ended abnormally on shutdown");
802 }
803 }
804 }
805 Err(_) => tracing::warn!(
806 "scheduler + background pool did not drain within {SHUTDOWN_POOL_DRAIN_SECS}s of \
807 shutdown; proceeding (work is idempotent)"
808 ),
809 }
810
811 tracing::info!("Server shut down gracefully");
812 }
813
814 /// Filesystem path to the business-assumptions TOML, `ASSUMPTIONS_PATH` or its
815 /// default. Single source for the default so every reader agrees.
816 /// Shutdown budget, reconciled so the background-pool drain can't be truncated
817 /// by the hard-exit deadline. The hard-exit timer arms at
818 /// signal receipt and must outlast the in-flight request drain PLUS the
819 /// post-serve drain window, otherwise the later drain is cut off mid-flight.
820 /// The two post-serve drains run *concurrently* (see `join!` below), so they
821 /// share the pool-drain window rather than summing to twice it.
822 const SHUTDOWN_REQUEST_DRAIN_SECS: u64 = 10;
823 const SHUTDOWN_POOL_DRAIN_SECS: u64 = 5;
824 /// Hard-exit backstop, defined as its own ceiling (not the sum of the two
825 /// drains) with margin above them, the process is force-killed only if BOTH
826 /// bounded drains overrun their budgets. The margin also makes the invariant
827 /// below a real check rather than the tautology it was when this equalled the
828 /// sum.
829 const SHUTDOWN_HARD_EXIT_SECS: u64 = 20;
830 const _: () = assert!(
831 SHUTDOWN_HARD_EXIT_SECS > SHUTDOWN_REQUEST_DRAIN_SECS + SHUTDOWN_POOL_DRAIN_SECS,
832 "hard-exit must outlast the bounded request + pool drains, with margin"
833 );
834
835 /// What an inspection flag asks for. Both answers are printable without a
836 /// database, an env file, or anything else the deployed environment supplies.
837 #[derive(Debug, PartialEq, Eq)]
838 enum Inspection {
839 Version,
840 Help,
841 }
842
843 /// Classify the arguments, without printing. Unknown arguments are left alone:
844 /// the binary takes other flags (`--seed-examples`) later in startup, and this
845 /// pass is not an argument parser.
846 fn inspection_request<I, S>(args: I) -> Option<Inspection>
847 where
848 I: IntoIterator<Item = S>,
849 S: AsRef<str>,
850 {
851 args.into_iter().find_map(|arg| match arg.as_ref() {
852 "--version" | "-V" => Some(Inspection::Version),
853 "--help" | "-h" => Some(Inspection::Help),
854 _ => None,
855 })
856 }
857
858 /// `name version (git sha)`, the same string `/health` reports, so the two can
859 /// be compared directly when verifying a release.
860 fn version_line() -> String {
861 match option_env!("GIT_HASH") {
862 Some(hash) if !hash.is_empty() => {
863 format!("makenotwork {} ({hash})", env!("CARGO_PKG_VERSION"))
864 }
865 _ => format!("makenotwork {}", env!("CARGO_PKG_VERSION")),
866 }
867 }
868
869 const HELP: &str = "\
870 makenotwork, the MNW server.
871
872 Usage: makenotwork [FLAGS]
873
874 Flags:
875 -V, --version Print the version and exit
876 -h, --help Print this help and exit
877 --seed-examples Seed the staging example marketplace, then exit. Refuses
878 against production; see makenotwork::seed.
879
880 The server takes its configuration from the environment (see .env.example).
881 Deployment tooling drives these modes by environment variable rather than by
882 flag, because each one runs against a target with its env already sourced:
883
884 MNW_CHECK_CONFIG=1 Load the config, print a sentinel line, exit. No database.
885 MNW_CHECK_DOCS=1 Report broken internal doc links, exit. No database.
886 SANDO_BOOT_SMOKE=1 Serve a minimal /health on SANDO_BOOT_SMOKE_PORT and idle.
887
888 Running with no flags starts the server: config, pool, migrations, then serve.";
889
890 /// Answer `--version` and `--help` before the process touches its environment.
891 /// Returns the exit code when a flag was handled, `None` to carry on booting.
892 fn handle_inspection_flags() -> Option<i32> {
893 match inspection_request(std::env::args().skip(1))? {
894 Inspection::Version => println!("{}", version_line()),
895 Inspection::Help => println!("{HELP}"),
896 }
897 Some(0)
898 }
899
900 /// Wait for SIGINT (Ctrl-C) or SIGTERM, then return to trigger graceful shutdown.
901 /// In-flight requests get up to `SHUTDOWN_REQUEST_DRAIN_SECS` to complete, then
902 /// the post-serve drains get `SHUTDOWN_POOL_DRAIN_SECS`; the hard-exit timer
903 /// (`SHUTDOWN_HARD_EXIT_SECS`) is the backstop that frees a hung restart.
904 async fn shutdown_signal() {
905 use tokio::signal;
906
907 let ctrl_c = async {
908 signal::ctrl_c()
909 .await
910 .expect("Failed to install Ctrl+C handler");
911 };
912
913 #[cfg(unix)]
914 let terminate = async {
915 signal::unix::signal(signal::unix::SignalKind::terminate())
916 .expect("Failed to install SIGTERM handler")
917 .recv()
918 .await;
919 };
920
921 #[cfg(not(unix))]
922 let terminate = std::future::pending::<()>();
923
924 tokio::select! {
925 () = ctrl_c => tracing::info!("Received SIGINT, starting graceful shutdown"),
926 () = terminate => tracing::info!("Received SIGTERM, starting graceful shutdown"),
927 }
928
929 // Spawn the hard-exit backstop: if the full graceful drain (request drain +
930 // concurrent pool drain) overruns its reconciled budget, force exit so a hung
931 // connection can't block a deployment restart indefinitely.
932 tokio::spawn(async {
933 tokio::time::sleep(Duration::from_secs(SHUTDOWN_HARD_EXIT_SECS)).await;
934 eprintln!("Graceful shutdown timed out after {SHUTDOWN_HARD_EXIT_SECS}s, forcing exit");
935 std::process::exit(1);
936 });
937 }
938
939 #[cfg(test)]
940 mod inspection_tests {
941 use super::*;
942
943 #[test]
944 fn version_flags_are_recognised() {
945 assert_eq!(inspection_request(["--version"]), Some(Inspection::Version));
946 assert_eq!(inspection_request(["-V"]), Some(Inspection::Version));
947 }
948
949 #[test]
950 fn help_flags_are_recognised() {
951 assert_eq!(inspection_request(["--help"]), Some(Inspection::Help));
952 assert_eq!(inspection_request(["-h"]), Some(Inspection::Help));
953 }
954
955 #[test]
956 fn boot_arguments_are_not_inspection_requests() {
957 assert_eq!(inspection_request(Vec::<String>::new()), None);
958 assert_eq!(inspection_request(["--seed-examples"]), None);
959 }
960
961 #[test]
962 fn the_first_inspection_flag_wins() {
963 assert_eq!(
964 inspection_request(["--seed-examples", "--version", "--help"]),
965 Some(Inspection::Version)
966 );
967 }
968
969 #[test]
970 fn version_line_carries_the_crate_version() {
971 assert!(version_line().starts_with(&format!("makenotwork {}", env!("CARGO_PKG_VERSION"))));
972 }
973 }
974