Skip to main content

max / makenotwork

19.1 KB · 461 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, Instant};
7 use tower_http::request_id::{MakeRequestUuid, PropagateRequestIdLayer, SetRequestIdLayer};
8 use tower_http::trace::TraceLayer;
9 use tower_sessions::cookie::time::Duration as CookieDuration;
10 use tower_sessions::cookie::SameSite;
11 use tower_sessions::{Expiry, SessionManagerLayer};
12 use tower_sessions_sqlx_store::PostgresStore;
13 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt, Layer};
14
15 use makenotwork::config::Config;
16 use makenotwork::constants;
17 use docengine::{Assumptions, DocLoader, DocLoaderConfig};
18 use makenotwork::email::{EmailClient, EmailConfig};
19 use makenotwork::payments::StripeClient;
20 use makenotwork::storage::S3Client;
21 use makenotwork::scanning::ScanPipeline;
22 use makenotwork::{build_app, AppState};
23 use webauthn_rs::WebauthnBuilder;
24
25 #[tokio::main]
26 async fn main() {
27 dotenvy::dotenv().ok();
28
29 // JSON in release, human-readable in dev
30 tracing_subscriber::registry()
31 .with(
32 tracing_subscriber::EnvFilter::try_from_default_env()
33 .unwrap_or_else(|_| "makenotwork=debug,tower_http=debug,sqlx=info".into()),
34 )
35 .with(if cfg!(debug_assertions) {
36 tracing_subscriber::fmt::layer().boxed()
37 } else {
38 tracing_subscriber::fmt::layer().json().boxed()
39 })
40 .init();
41
42 let config = Config::from_env().expect("Failed to load configuration");
43 tracing::info!("Configuration loaded");
44
45 // Create database connection pool with health checks and lifecycle limits.
46 // - test_before_acquire: validates connections before use (catches stale/broken conns)
47 // - max_lifetime: rotates connections to prevent long-lived session issues
48 // - idle_timeout: prunes idle connections to free server resources
49 // - min_connections: keeps warm connections ready for immediate use
50 // - log_slow_statements: logs queries exceeding 100ms at WARN level
51 let connect_options: sqlx::postgres::PgConnectOptions = config.database_url.parse()
52 .expect("Invalid DATABASE_URL");
53 let connect_options = connect_options
54 .log_statements(log::LevelFilter::Trace)
55 .log_slow_statements(log::LevelFilter::Warn, Duration::from_millis(100));
56
57 let db = PgPoolOptions::new()
58 .max_connections(constants::DB_POOL_MAX_CONNECTIONS)
59 .min_connections(constants::DB_POOL_MIN_CONNECTIONS)
60 .acquire_timeout(Duration::from_secs(constants::DB_ACQUIRE_TIMEOUT_SECS))
61 .max_lifetime(Duration::from_secs(constants::DB_MAX_LIFETIME_SECS))
62 .idle_timeout(Duration::from_secs(constants::DB_IDLE_TIMEOUT_SECS))
63 .test_before_acquire(true)
64 .connect_with(connect_options)
65 .await
66 .expect("Failed to connect to database");
67
68 tracing::info!("Database connected");
69
70 // Run migrations — exit cleanly (code 2) on failure instead of panicking.
71 // This prevents systemd from crash-looping on migration errors (e.g. version
72 // conflicts after a bad deploy). WAM ticket alerts the operator.
73 if let Err(e) = sqlx::migrate!("./migrations").run(&db).await {
74 tracing::error!(error = %e, "Migration failed — exiting without restart");
75
76 // Best-effort WAM alert (DB is up, WAM may be reachable)
77 if let Ok(wam_url) = std::env::var("WAM_URL") {
78 let body = format!("Migration error on startup:\n{e}");
79 let _ = reqwest::Client::new()
80 .post(format!("{wam_url}/tickets"))
81 .json(&serde_json::json!({
82 "title": "Migration failure — server not starting",
83 "body": body,
84 "priority": "critical",
85 "source": "migration-failure",
86 }))
87 .timeout(std::time::Duration::from_secs(5))
88 .send()
89 .await;
90 }
91
92 // Exit code 2 — systemd configured not to restart on this code
93 std::process::exit(2);
94 }
95
96 tracing::info!("Migrations complete");
97
98 // Create PostgreSQL-backed session store (persists across restarts)
99 let session_store = PostgresStore::new(db.clone());
100 session_store
101 .migrate()
102 .await
103 .expect("Failed to migrate session store");
104
105 // In release mode, require HTTPS for session cookies (override with INSECURE_COOKIES=1 for staging)
106 let secure_cookies =
107 !cfg!(debug_assertions) && std::env::var("INSECURE_COOKIES").unwrap_or_default() != "1";
108 if secure_cookies {
109 tracing::info!("Session cookies configured for HTTPS (secure=true)");
110 } else if !cfg!(debug_assertions) {
111 tracing::warn!("Session cookies set to insecure (INSECURE_COOKIES=1)");
112 }
113
114 let session_layer = SessionManagerLayer::new(session_store)
115 .with_secure(secure_cookies)
116 .with_http_only(true)
117 .with_same_site(SameSite::Lax) // Lax allows session on top-level navigations (OAuth redirects)
118 .with_expiry(Expiry::OnInactivity(CookieDuration::days(
119 constants::SESSION_EXPIRY_DAYS,
120 )));
121
122 tracing::info!("PostgreSQL session store initialized");
123
124 // Initialize S3 client if storage is configured
125 let s3: Option<std::sync::Arc<dyn makenotwork::storage::StorageBackend>> =
126 if let Some(ref storage_config) = config.storage {
127 match S3Client::new(storage_config, &config.host_url).await {
128 Ok(client) => {
129 tracing::info!(bucket = %storage_config.bucket, "S3 storage initialized");
130 Some(std::sync::Arc::new(client))
131 }
132 Err(e) => {
133 tracing::warn!(error = ?e, "failed to initialize S3 storage, file uploads unavailable");
134 None
135 }
136 }
137 } else {
138 tracing::info!("S3 storage not configured. File uploads will be unavailable.");
139 None
140 };
141
142 // Initialize SyncKit blob S3 client if configured (separate bucket)
143 let synckit_s3: Option<std::sync::Arc<dyn makenotwork::storage::StorageBackend>> =
144 if let Some(ref synckit_storage_config) = config.synckit_storage {
145 match S3Client::new(synckit_storage_config, &config.host_url).await {
146 Ok(client) => {
147 tracing::info!(bucket = %synckit_storage_config.bucket, "SyncKit blob storage initialized");
148 Some(std::sync::Arc::new(client))
149 }
150 Err(e) => {
151 tracing::warn!(error = ?e, "Failed to initialize SyncKit blob storage");
152 None
153 }
154 }
155 } else {
156 tracing::info!("SyncKit blob storage not configured");
157 None
158 };
159
160 // Initialize Stripe client if configured
161 let stripe: Option<std::sync::Arc<dyn makenotwork::payments::PaymentProvider>> = if let Some(ref stripe_config) = config.stripe {
162 tracing::info!("Stripe payments initialized");
163 Some(std::sync::Arc::new(StripeClient::new(stripe_config)))
164 } else {
165 tracing::info!("Stripe not configured. Payments will be unavailable.");
166 None
167 };
168
169 // Initialize email client (logs in dev mode when POSTMARK_TOKEN is not set)
170 let email = EmailClient::new(EmailConfig::from_env(), Some(db.clone()));
171
172 // Load business assumptions (single source of truth for figures in the docs).
173 // Validation failure aborts startup — we don't want to serve stale numbers.
174 let assumptions_path = std::env::var("ASSUMPTIONS_PATH")
175 .unwrap_or_else(|_| "docs/internal/business/assumptions.toml".to_string());
176 let assumptions = std::sync::Arc::new(
177 match Assumptions::load(&assumptions_path) {
178 Ok(a) => {
179 if let Err(e) = a.validate() {
180 panic!("assumptions validation failed:\n{e}");
181 }
182 tracing::info!(path = %assumptions_path, "assumptions loaded and validated");
183 a
184 }
185 Err(e) => panic!("failed to load assumptions from {assumptions_path}: {e}"),
186 }
187 );
188
189 // Load documentation pages from disk
190 let docs_path = std::env::var("DOCS_PATH").unwrap_or_else(|_| "site-docs/public".to_string());
191 let docs = std::sync::Arc::new(DocLoader::load(
192 std::path::Path::new(&docs_path),
193 &DocLoaderConfig {
194 sections: vec![
195 ("about".to_string(), "About".to_string()),
196 ("guide".to_string(), "Guide".to_string()),
197 ("developer".to_string(), "Developer".to_string()),
198 ("legal".to_string(), "Legal".to_string()),
199 ("support".to_string(), "Support".to_string()),
200 ("tech".to_string(), "Tech".to_string()),
201 ],
202 link_prefix: "/docs".to_string(),
203 unpublished_pattern: Some("unpublished/".to_string()),
204 examples_path: Some(std::path::Path::new(&docs_path).join("../examples")),
205 pre_process: {
206 let assumptions = assumptions.clone();
207 Some(Box::new(move |md: &str| {
208 assumptions.substitute(md).map_err(|e| e.to_string())
209 }))
210 },
211 },
212 ));
213
214 // Initialize file scanning pipeline (optional). If scanning is configured,
215 // assert at least one AV layer is actually live — otherwise the FailOpen
216 // policy on ClamAV silently passes everything as Clean and the deploy
217 // looks healthy while shipping zero real coverage. Refuse to boot.
218 let scanner = if let Some(ref scan_config) = config.scan {
219 match ScanPipeline::new(scan_config) {
220 Ok(s) => match s.assert_live().await {
221 Ok(()) => {
222 tracing::info!("File scanning enabled");
223 Some(std::sync::Arc::new(s))
224 }
225 Err(e) => panic!("Scanning configured but no live AV layer: {e}"),
226 },
227 Err(e) => {
228 tracing::warn!(error = %e, "File scanning disabled");
229 None
230 }
231 }
232 } else {
233 tracing::info!("File scanning not configured");
234 None
235 };
236
237 // Initialize WebAuthn from HOST_URL (passkey / passwordless login)
238 let rp_origin = url::Url::parse(&config.host_url).expect("HOST_URL is not a valid URL");
239 let rp_id = rp_origin.host_str().expect("HOST_URL has no host").to_string();
240 let webauthn = std::sync::Arc::new(
241 WebauthnBuilder::new(&rp_id, &rp_origin)
242 .expect("Failed to create WebauthnBuilder")
243 .rp_name("Makenotwork")
244 .build()
245 .expect("Failed to build Webauthn"),
246 );
247 tracing::info!("WebAuthn initialized (rp_id={rp_id})");
248
249 // Initialize syntax highlighter if git repos path is configured
250 let syntax = if config.git_repos_path.is_some() {
251 tracing::info!(path = ?config.git_repos_path, "Git source browser enabled");
252 Some(std::sync::Arc::new(makenotwork::git::SyntaxHighlighter::new()))
253 } else {
254 tracing::info!("Git source browser not configured (GIT_REPOS_PATH unset)");
255 None
256 };
257
258 // Construct MT client when both mt_base_url and internal_shared_secret are set
259 let mt_client = match (&config.mt_base_url, &config.internal_shared_secret) {
260 (Some(base_url), Some(secret)) => {
261 tracing::info!(base_url = %base_url, "MT integration enabled");
262 Some(makenotwork::mt_client::MtClient::new(base_url.clone(), secret.clone()))
263 }
264 _ => {
265 tracing::info!("MT integration not configured (need MT_BASE_URL + INTERNAL_SHARED_SECRET)");
266 None
267 }
268 };
269
270 // WAM ticket manager client (tailnet-only, for operational alerts)
271 let wam = config.wam_url.as_ref().map(|url| {
272 tracing::info!(url = %url, "WAM integration enabled");
273 makenotwork::wam_client::WamClient::new(url.clone())
274 });
275
276 // Warm custom domain cache
277 let domain_cache = std::sync::Arc::new(dashmap::DashMap::new());
278 match makenotwork::db::custom_domains::get_all_verified_domains(&db).await {
279 Ok(domains) => {
280 for d in &domains {
281 domain_cache.insert(d.domain.clone(), d.user_id);
282 }
283 if !domains.is_empty() {
284 tracing::info!(count = domains.len(), "Custom domain cache warmed");
285 }
286 }
287 Err(e) => {
288 tracing::warn!(error = ?e, "Failed to warm custom domain cache");
289 }
290 }
291
292 let started_at = chrono::Utc::now();
293 let start_instant = Instant::now();
294 let page_view_tx = makenotwork::db::page_views::spawn_batcher(db.clone());
295 let bg = makenotwork::background::spawn_pool();
296 let state = AppState {
297 db,
298 config: config.clone(),
299 s3,
300 synckit_s3,
301 stripe,
302 email,
303 docs,
304 tier_prices: makenotwork::tier_prices::TierPrices::from_assumptions(&assumptions),
305 scanner,
306 webauthn,
307 syntax,
308 started_at,
309 start_instant,
310 session_cache: std::sync::Arc::new(dashmap::DashMap::new()),
311 mt_client,
312 wam,
313 domain_cache,
314 scan_semaphore: std::sync::Arc::new(tokio::sync::Semaphore::new(makenotwork::constants::SCAN_MAX_CONCURRENT)),
315 caddy_ask_semaphore: std::sync::Arc::new(tokio::sync::Semaphore::new(makenotwork::constants::CADDY_ASK_MAX_CONCURRENT)),
316 restart_at: std::sync::Arc::new(std::sync::atomic::AtomicI64::new(0)),
317 sync_notify: std::sync::Arc::new(dashmap::DashMap::new()),
318 sse_connections: std::sync::Arc::new(dashmap::DashMap::new()),
319 metrics_handle: Some(makenotwork::metrics::init()),
320 page_view_tx,
321 bg,
322 };
323
324 // Log active features at startup
325 tracing::info!(
326 s3 = state.s3.is_some(),
327 synckit_s3 = state.synckit_s3.is_some(),
328 stripe = state.stripe.is_some(),
329 scanner = state.scanner.is_some(),
330 mt = state.mt_client.is_some(),
331 wam = state.wam.is_some(),
332 git = state.config.git_repos_path.is_some(),
333 "Active features"
334 );
335
336 // Start background health monitor and scheduler
337 let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(());
338 let _monitor_handle = makenotwork::monitor::spawn_monitor(state.clone(), shutdown_rx);
339 let scheduler_shutdown_rx = shutdown_tx.subscribe();
340 let _scheduler_handle = makenotwork::scheduler::spawn_scheduler(state.clone(), scheduler_shutdown_rx);
341
342 // Start scan worker pool. Only meaningful if a scanner is configured;
343 // otherwise enqueue_scan_for never enqueues (trust-gate fast path).
344 if let (Some(scanner), Some(s3_for_workers)) = (state.scanner.clone(), state.s3.clone()) {
345 let scan_ctx = std::sync::Arc::new(makenotwork::scanning::worker::WorkerContext {
346 db: state.db.clone(),
347 s3: s3_for_workers,
348 pipeline: scanner,
349 scan_semaphore: state.scan_semaphore.clone(),
350 wam: state.wam.clone(),
351 });
352 let worker_count = makenotwork::constants::SCAN_WORKER_COUNT;
353 let worker_shutdown_rx = shutdown_tx.subscribe();
354 makenotwork::scanning::worker::spawn_pool(worker_count, scan_ctx, worker_shutdown_rx);
355 tracing::info!(worker_count, "scan worker pool started");
356
357 let report = makenotwork::scanning::spool::reap_all(
358 std::path::Path::new(makenotwork::constants::SCAN_SPOOL_DIR),
359 );
360 if report.deleted > 0 || report.errors > 0 {
361 tracing::info!(
362 deleted = report.deleted,
363 errors = report.errors,
364 "scan spool startup reaper completed"
365 );
366 }
367 }
368
369 // Build router (shared with integration tests via lib.rs)
370 let app = build_app(state, session_layer)
371 // Request ID: propagate → trace → set (Axum applies inside-out)
372 .layer(PropagateRequestIdLayer::x_request_id())
373 .layer(
374 TraceLayer::new_for_http()
375 .make_span_with(|request: &Request<_>| {
376 let request_id = request
377 .headers()
378 .get("x-request-id")
379 .and_then(|v| v.to_str().ok())
380 .unwrap_or("-");
381
382 tracing::info_span!(
383 "request",
384 method = %request.method(),
385 uri = %request.uri(),
386 request_id = %request_id,
387 user_id = tracing::field::Empty,
388 )
389 })
390 .on_response(|response: &axum::http::Response<_>, latency: Duration, _span: &tracing::Span| {
391 let status = response.status().as_u16();
392 if status >= 500 {
393 tracing::error!(status, latency_ms = latency.as_millis() as u64, "response");
394 } else if status >= 400 {
395 tracing::warn!(status, latency_ms = latency.as_millis() as u64, "response");
396 } else {
397 tracing::debug!(status, latency_ms = latency.as_millis() as u64, "response");
398 }
399 }),
400 )
401 .layer(SetRequestIdLayer::x_request_id(MakeRequestUuid));
402
403 let addr = config.socket_addr();
404 tracing::info!(addr = %addr, "listening");
405
406 let listener = tokio::net::TcpListener::bind(addr)
407 .await
408 .expect("Failed to bind to address");
409
410 // Use into_make_service_with_connect_info to provide peer IP for rate limiting.
411 // After the shutdown signal fires, in-flight requests get a 10-second window
412 // to complete before the process force-exits. This prevents hung connections
413 // from blocking deployment restarts indefinitely.
414 axum::serve(
415 listener,
416 app.into_make_service_with_connect_info::<std::net::SocketAddr>(),
417 )
418 .with_graceful_shutdown(shutdown_signal())
419 .await
420 .expect("Server error");
421
422 drop(shutdown_tx); // signal monitor to stop
423 tracing::info!("Server shut down gracefully");
424 }
425
426 /// Wait for SIGINT (Ctrl-C) or SIGTERM, then return to trigger graceful shutdown.
427 /// In-flight requests are allowed up to 10 seconds to complete. After that,
428 /// the process force-exits to prevent hung connections from blocking restarts.
429 async fn shutdown_signal() {
430 use tokio::signal;
431
432 let ctrl_c = async {
433 signal::ctrl_c()
434 .await
435 .expect("Failed to install Ctrl+C handler");
436 };
437
438 #[cfg(unix)]
439 let terminate = async {
440 signal::unix::signal(signal::unix::SignalKind::terminate())
441 .expect("Failed to install SIGTERM handler")
442 .recv()
443 .await;
444 };
445
446 #[cfg(not(unix))]
447 let terminate = std::future::pending::<()>();
448
449 tokio::select! {
450 () = ctrl_c => tracing::info!("Received SIGINT, starting graceful shutdown"),
451 () = terminate => tracing::info!("Received SIGTERM, starting graceful shutdown"),
452 }
453
454 // Spawn a hard deadline: if graceful drain takes longer than 10s, force exit
455 tokio::spawn(async {
456 tokio::time::sleep(Duration::from_secs(10)).await;
457 eprintln!("Graceful shutdown timed out after 10s, forcing exit");
458 std::process::exit(1);
459 });
460 }
461