Skip to main content

max / makenotwork

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