use std::{ path::Path, sync::{Arc, Mutex}, }; use axum::{ Json, Router, extract::{DefaultBodyLimit, State}, http::StatusCode, response::{IntoResponse, Response}, routing::get, }; use rusqlite::{Connection, params, types::Type}; use serde::{Deserialize, Serialize}; use tokio::{sync::Semaphore, task}; const MAX_HIGH_SCORES: usize = 10; const MAX_NAME_CHARS: usize = 21; const MAX_SUBMISSION_BODY_BYTES: usize = 1024; const MAX_CONCURRENT_DATABASE_OPERATIONS: usize = 1; const DEFAULT_HIGH_SCORES: &[(&str, u32)] = &[ ("Paul", 6_537_392), ("Paul Schulze", 2_979_000), ("Kalle", 2_393_464), ("Paul Schulze", 2_328_000), ("Martina Sommer", 2_326_000), ("Martina Sommer", 1_093_000), ("Sommer Martina", 1_027_000), ("No Name", 1_000_000), ("TDK Pinball Player", 923_000), ("Martina Sommer", 905_000), ]; #[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)] pub struct HighScore { pub name: String, pub score: u32, } #[derive(Debug)] pub enum StoreError { Database(rusqlite::Error), Poisoned, } impl From for StoreError { fn from(error: rusqlite::Error) -> Self { Self::Database(error) } } #[derive(Clone)] pub struct HighScoreStore(Arc>); impl HighScoreStore { /// Open or create a SQLite-backed high-score store. /// /// # Errors /// /// Returns the SQLite error raised while opening or initializing the /// database. pub fn open(path: impl AsRef) -> Result { let path = path.as_ref(); let seed_defaults = !path.exists(); Self::from_connection(Connection::open(path)?, seed_defaults) } /// Create an in-memory high-score store for tests or short-lived runs. /// /// # Errors /// /// Returns the SQLite error raised while initializing the database. pub fn open_in_memory() -> Result { Self::from_connection(Connection::open_in_memory()?, false) } fn from_connection( mut connection: Connection, seed_defaults: bool, ) -> Result { connection.execute_batch( "CREATE TABLE IF NOT EXISTS high_scores ( id INTEGER PRIMARY KEY, name TEXT NOT NULL, score INTEGER NOT NULL CHECK (score >= 0 AND score <= 4294967295) ); CREATE INDEX IF NOT EXISTS high_scores_order ON high_scores (score DESC, id ASC);", )?; if seed_defaults { let transaction = connection.transaction()?; for &(name, score) in DEFAULT_HIGH_SCORES { transaction.execute( "INSERT INTO high_scores (name, score) VALUES (?1, ?2)", params![name, i64::from(score)], )?; } transaction.commit()?; } Ok(Self(Arc::new(Mutex::new(connection)))) } /// Return the current table in descending score order. /// /// # Errors /// /// Returns an error if the database lock or query fails. pub fn list(&self) -> Result, StoreError> { let connection = self.0.lock().map_err(|_| StoreError::Poisoned)?; let mut statement = connection.prepare( "SELECT name, score FROM high_scores ORDER BY score DESC, id ASC LIMIT ?1", )?; let scores = statement .query_map([i64::try_from(MAX_HIGH_SCORES).unwrap_or(10)], |row| { let score: i64 = row.get(1)?; let score = u32::try_from(score).map_err(|error| { rusqlite::Error::FromSqlConversionFailure(1, Type::Integer, Box::new(error)) })?; Ok(HighScore { name: row.get(0)?, score, }) })? .collect::, _>>()?; Ok(scores) } /// Add one score and return the resulting canonical top ten. /// /// # Errors /// /// Returns an error if the database lock or transaction fails. pub fn submit(&self, entry: &HighScore) -> Result, StoreError> { { let mut connection = self.0.lock().map_err(|_| StoreError::Poisoned)?; let transaction = connection.transaction()?; transaction.execute( "INSERT INTO high_scores (name, score) VALUES (?1, ?2)", params![entry.name, i64::from(entry.score)], )?; transaction.execute( "DELETE FROM high_scores WHERE id NOT IN ( SELECT id FROM high_scores ORDER BY score DESC, id ASC LIMIT ?1 )", [i64::try_from(MAX_HIGH_SCORES).unwrap_or(10)], )?; transaction.commit()?; } self.list() } } #[derive(Clone)] struct AppState { store: HighScoreStore, database_slots: Arc, } impl AppState { fn new(store: HighScoreStore) -> Self { Self { store, database_slots: Arc::new(Semaphore::new(MAX_CONCURRENT_DATABASE_OPERATIONS)), } } } #[derive(Debug)] enum DatabaseRequestError { Busy, Failed, } async fn run_database_operation( state: &AppState, operation: F, ) -> Result where T: Send + 'static, F: FnOnce(&HighScoreStore) -> Result + Send + 'static, { let permit = state .database_slots .clone() .try_acquire_owned() .map_err(|_| DatabaseRequestError::Busy)?; let store = state.store.clone(); task::spawn_blocking(move || { let _permit = permit; operation(&store) }) .await .map_err(|_| DatabaseRequestError::Failed)? .map_err(|_| DatabaseRequestError::Failed) } #[derive(Debug, Deserialize)] struct SubmitRequest { name: String, score: u32, } #[derive(Debug, Serialize)] struct ErrorResponse { error: &'static str, } fn invalid_request(message: &'static str) -> Response { ( StatusCode::BAD_REQUEST, Json(ErrorResponse { error: message }), ) .into_response() } fn database_error_response(error: &DatabaseRequestError) -> Response { match error { DatabaseRequestError::Busy => ( StatusCode::SERVICE_UNAVAILABLE, Json(ErrorResponse { error: "service is busy", }), ) .into_response(), DatabaseRequestError::Failed => StatusCode::INTERNAL_SERVER_ERROR.into_response(), } } fn validate_request(request: &SubmitRequest) -> Result { let name = request.name.trim(); if name.is_empty() { return Err("name must not be empty"); } if name.chars().count() > MAX_NAME_CHARS { return Err("name is too long"); } if name.chars().any(char::is_control) { return Err("name contains a control character"); } Ok(HighScore { name: name.to_owned(), score: request.score, }) } async fn health() -> &'static str { "ok" } async fn list_high_scores(State(state): State) -> Response { match run_database_operation(&state, HighScoreStore::list).await { Ok(scores) => Json(scores).into_response(), Err(error) => database_error_response(&error), } } async fn submit_high_score( State(state): State, Json(request): Json, ) -> Response { let entry = match validate_request(&request) { Ok(entry) => entry, Err(message) => return invalid_request(message), }; match run_database_operation(&state, move |store| store.submit(&entry)).await { Ok(scores) => (StatusCode::CREATED, Json(scores)).into_response(), Err(error) => database_error_response(&error), } } pub fn router(store: HighScoreStore) -> Router { router_with_state(AppState::new(store)) } fn router_with_state(state: AppState) -> Router { Router::new() .route("/healthz", get(health)) .route( "/api/highscores", get(list_high_scores) .post(submit_high_score) .layer(DefaultBodyLimit::max(MAX_SUBMISSION_BODY_BYTES)), ) .with_state(state) } #[cfg(test)] mod tests { use std::{sync::mpsc, time::Duration}; use axum::{ body::Body, http::{Request, StatusCode}, }; use http_body_util::BodyExt; use tempfile::tempdir; use tower::ServiceExt; use super::*; async fn submit(app: &Router, name: &str, score: u32) -> StatusCode { app.clone() .oneshot( Request::post("/api/highscores") .header("content-type", "application/json") .body(Body::from( serde_json::to_vec(&serde_json::json!({ "name": name, "score": score, })) .expect("request JSON should encode"), )) .expect("request should build"), ) .await .expect("router should respond") .status() } async fn request(app: &Router, request: Request) -> StatusCode { app.clone() .oneshot(request) .await .expect("router should respond") .status() } fn hold_database_lock( store: HighScoreStore, ) -> (mpsc::SyncSender<()>, std::thread::JoinHandle<()>) { let (locked_sender, locked_receiver) = mpsc::sync_channel(0); let (release_sender, release_receiver) = mpsc::sync_channel(0); let lock_thread = std::thread::spawn(move || { let _connection = store.0.lock().expect("database lock should succeed"); locked_sender .send(()) .expect("test should observe the held database lock"); release_receiver .recv() .expect("test should release the database lock"); }); locked_receiver .recv() .expect("database lock thread should start"); (release_sender, lock_thread) } async fn wait_until_database_is_busy(database_slots: &Semaphore) { tokio::time::timeout(Duration::from_secs(1), async { while database_slots.available_permits() != 0 { tokio::task::yield_now().await; } }) .await .expect("first database request should acquire admission"); } async fn list(app: &Router) -> Vec { let response = app .clone() .oneshot( Request::get("/api/highscores") .body(Body::empty()) .expect("request should build"), ) .await .expect("router should respond"); assert_eq!(response.status(), StatusCode::OK); let body = response .into_body() .collect() .await .expect("response body should read") .to_bytes(); serde_json::from_slice(&body).expect("response should contain scores") } #[tokio::test] async fn api_persists_and_keeps_the_top_ten() { let directory = tempdir().expect("temporary directory should exist"); let database = directory.path().join("highscores.sqlite3"); std::fs::File::create(&database).expect("database file should exist"); let store = HighScoreStore::open(&database).expect("database should open"); let app = router(store); for score in 0..12 { assert_eq!( submit(&app, &format!("Player {score}"), score).await, StatusCode::CREATED ); } let scores = list(&app).await; assert_eq!(scores.len(), MAX_HIGH_SCORES); assert_eq!(scores[0].score, 11); assert_eq!(scores[9].score, 2); drop(app); let reopened = router(HighScoreStore::open(&database).expect("database should reopen")); assert_eq!(list(&reopened).await, scores); } #[test] fn new_database_starts_with_the_original_demo_scores() { let directory = tempdir().expect("temporary directory should exist"); let database = directory.path().join("highscores.sqlite3"); let store = HighScoreStore::open(&database).expect("database should open"); let scores = store.list().expect("scores should list"); let expected = DEFAULT_HIGH_SCORES .iter() .map(|&(name, score)| HighScore { name: name.to_owned(), score, }) .collect::>(); assert_eq!(scores, expected); } #[test] fn existing_database_is_not_seeded() { let directory = tempdir().expect("temporary directory should exist"); let database = directory.path().join("highscores.sqlite3"); std::fs::File::create(&database).expect("database file should exist"); let store = HighScoreStore::open(&database).expect("database should open"); assert!(store.list().expect("scores should list").is_empty()); } #[tokio::test] async fn api_rejects_invalid_names() { let app = router(HighScoreStore::open_in_memory().expect("database should open")); assert_eq!(submit(&app, "", 10).await, StatusCode::BAD_REQUEST); assert_eq!( submit(&app, "abcdefghijklmnopqrstuv", 10).await, StatusCode::BAD_REQUEST ); assert_eq!(submit(&app, "ok\nno", 10).await, StatusCode::BAD_REQUEST); } #[tokio::test] async fn api_rejects_oversized_submission_bodies() { let app = router(HighScoreStore::open_in_memory().expect("database should open")); let body = serde_json::to_vec(&serde_json::json!({ "name": "Player", "score": 10, "padding": "x".repeat(MAX_SUBMISSION_BODY_BYTES), })) .expect("request JSON should encode"); let status = request( &app, Request::post("/api/highscores") .header("content-type", "application/json") .body(Body::from(body)) .expect("request should build"), ) .await; assert_eq!(status, StatusCode::PAYLOAD_TOO_LARGE); } #[tokio::test(flavor = "multi_thread", worker_threads = 1)] async fn busy_database_load_sheds_api_without_blocking_health() { let store = HighScoreStore::open_in_memory().expect("database should open"); let (release_sender, lock_thread) = hold_database_lock(store.clone()); let state = AppState::new(store); let database_slots = state.database_slots.clone(); let app = router_with_state(state); let accepted_app = app.clone(); let accepted = tokio::spawn(async move { submit(&accepted_app, "Player", 10).await }); wait_until_database_is_busy(&database_slots).await; assert_eq!( tokio::time::timeout( Duration::from_millis(250), request( &app, Request::get("/healthz") .body(Body::empty()) .expect("request should build"), ), ) .await .expect("health request should not wait for the database"), StatusCode::OK ); assert_eq!( submit(&app, "Other Player", 20).await, StatusCode::SERVICE_UNAVAILABLE ); assert_eq!( request( &app, Request::get("/api/highscores") .body(Body::empty()) .expect("request should build"), ) .await, StatusCode::SERVICE_UNAVAILABLE ); assert_eq!( request( &app, Request::builder() .method("HEAD") .uri("/api/highscores") .body(Body::empty()) .expect("request should build"), ) .await, StatusCode::SERVICE_UNAVAILABLE ); release_sender .send(()) .expect("database lock should be released"); assert_eq!( accepted.await.expect("accepted request should complete"), StatusCode::CREATED ); lock_thread .join() .expect("database lock thread should stop"); } #[tokio::test(flavor = "multi_thread", worker_threads = 1)] async fn canceled_request_holds_admission_until_blocking_work_stops() { let store = HighScoreStore::open_in_memory().expect("database should open"); let (release_sender, lock_thread) = hold_database_lock(store.clone()); let state = AppState::new(store); let database_slots = state.database_slots.clone(); let app = router_with_state(state); let canceled_app = app.clone(); let canceled = tokio::spawn(async move { submit(&canceled_app, "Player", 10).await }); wait_until_database_is_busy(&database_slots).await; canceled.abort(); assert!( canceled .await .expect_err("request should be canceled") .is_cancelled() ); assert_eq!(database_slots.available_permits(), 0); assert_eq!( submit(&app, "Other Player", 20).await, StatusCode::SERVICE_UNAVAILABLE ); release_sender .send(()) .expect("database lock should be released"); lock_thread .join() .expect("database lock thread should stop"); tokio::time::timeout(Duration::from_secs(1), async { while database_slots.available_permits() == 0 { tokio::task::yield_now().await; } }) .await .expect("completed blocking work should release admission"); assert_eq!(submit(&app, "Other Player", 20).await, StatusCode::CREATED); } #[tokio::test] async fn health_endpoint_is_available_for_nginx() { let app = router(HighScoreStore::open_in_memory().expect("database should open")); let response = app .oneshot( Request::get("/healthz") .body(Body::empty()) .expect("request should build"), ) .await .expect("router should respond"); assert_eq!(response.status(), StatusCode::OK); } }