add(http): custom error type for http request

Cargo.lock

@@ -2866,7 +2866,7 @@ dependencies = [
 
 [[package]]
 name = "server"
-version = "0.2.0"
+version = "0.3.0"
 dependencies = [
  "argon2",
  "async-broadcast",
@@ -2887,6 +2887,7 @@ dependencies = [
  "serde_json",
  "str0m",
  "symphonia",
+ "thiserror 2.0.18",
  "tokio",
  "tower-http",
  "tracing",

crates/server/Cargo.toml

@@ -36,3 +36,4 @@ opus = "0.3.1"
 rubato = "3.0.0"
 chrono = "0.4.45"
 regex = "1"
+thiserror = "2.0.18"

crates/server/src/http.rs

@@ -46,6 +46,7 @@ use tracing::{debug, info, warn};
 use crate::{
     AppState,
     hash::{hash_password, verify_password},
+    http_error::HttpError,
     webrtc_ingest::handle_whip_injest,
 };
 
@@ -85,11 +86,6 @@ impl HttpServer {
             .allow_credentials(true)
             .allow_origin(origins);
 
-        // TODO: Add an error type that impls IntoResponse, enum HttpError {}; impl IntoResponse for
-        // HttpError
-        // https://docs.rs/thiserror/latest/thiserror/
-        //
-        // Its rather shitty
         let app = Router::new()
             .route("/api/catalog", get(catalog_handler))
             .route("/api/user", post(create_user_handler))
@@ -136,42 +132,38 @@ async fn edit_stream_key(
     State(state): State<Arc<HttpServer>>,
     auth: AuthUser,
     Json(payload): Json<EditStreamKeyRequest>,
-) -> impl IntoResponse {
+) -> Result<impl IntoResponse, HttpError> {
     // Trim surrounding whitespace so labels aren't stored with leading/trailing spaces.
     let new_label = payload.new.trim();
     // Length (1..=67) and allowed charset ([A-Za-z0-9 _-], space included).
     if !KEY_RE.is_match(new_label) {
-        return (StatusCode::BAD_REQUEST, "invalid label").into_response();
+        return Err(HttpError::BadRequest("invalid label".into()));
     }
     // Replaces the JS lookahead: label must contain at least one letter.
     if !new_label.chars().any(|c| c.is_ascii_alphabetic()) {
-        return (StatusCode::BAD_REQUEST, "label must contain a letter").into_response();
+        return Err(HttpError::BadRequest("label must contain a letter".into()));
     }
 
-    let stream_key_id_result = stream_key::Entity::find_by_id(payload.id)
+    let stream_key = stream_key::Entity::find_by_id(payload.id)
         .one(&state.db)
-        .await;
+        .await?
-    if stream_key_id_result.is_err() {
+        .ok_or(HttpError::NotFound)?;
-        return StatusCode::INTERNAL_SERVER_ERROR.into_response();
-    }
-    let stream_key = stream_key_id_result.unwrap().unwrap();
     // Do they own the stream key
     if stream_key.user_id != auth.0.id {
-        return StatusCode::FORBIDDEN.into_response();
+        return Err(HttpError::Forbidden);
     }
     stream_key
         .into_active_model()
         .change_label_value(&state.db, new_label.to_string())
-        .await
+        .await?;
-        .unwrap();
 
-    StatusCode::OK.into_response()
+    Ok(StatusCode::OK)
 }
 
-async fn catalog_handler(State(state): State<Arc<HttpServer>>) -> impl IntoResponse {
+async fn catalog_handler(
-    let streams = stream_session::Model::get_all_active_sessions(&state.db)
+    State(state): State<Arc<HttpServer>>,
-        .await
+) -> Result<impl IntoResponse, HttpError> {
-        .unwrap();
+    let streams = stream_session::Model::get_all_active_sessions(&state.db).await?;
     debug!("{:#?}", streams);
     let catalog: Vec<StreamListing> = futures::future::join_all(streams.iter().map(|listing| {
         let db = state.db.clone();
@@ -180,14 +172,12 @@ async fn catalog_handler(State(state): State<Arc<HttpServer>>) -> impl IntoRespo
         async move {
             let key_info = stream_key::Entity::find_by_id(stream_key_id)
                 .one(&db)
-                .await
+                .await?
-                .unwrap()
+                .ok_or(HttpError::NotFound)?;
-                .unwrap();
             let user = users::Entity::find_by_id(key_info.user_id)
                 .one(&db)
-                .await
+                .await?
-                .unwrap()
+                .ok_or(HttpError::NotFound)?;
-                .unwrap();
             let meow = state2
                 .appstate
                 .clone()
@@ -195,18 +185,20 @@ async fn catalog_handler(State(state): State<Arc<HttpServer>>) -> impl IntoRespo
                 .await
                 .stream_sessions
                 .get(&key_info.id)
-                .unwrap()
+                .ok_or(HttpError::NotFound)?
                 .started_at;
-            StreamListing {
+            Ok::<_, HttpError>(StreamListing {
                 id: stream_key_id,
                 label: key_info.label,
                 user: user.username,
                 started_at: meow,
-            }
+            })
         }
     }))
-    .await;
+    .await
-    Json(catalog)
+    .into_iter()
+    .collect::<Result<Vec<_>, HttpError>>()?;
+    Ok(Json(catalog))
 }
 
 #[derive(Deserialize)]
@@ -233,18 +225,17 @@ fn extract_session_token(headers: &HeaderMap) -> Option<String> {
 struct AuthUser(entity::users::Model);
 
 impl FromRequestParts<Arc<HttpServer>> for AuthUser {
-    type Rejection = StatusCode;
+    type Rejection = HttpError;
 
     async fn from_request_parts(
         parts: &mut Parts,
         state: &Arc<HttpServer>,
-    ) -> Result<Self, StatusCode> {
+    ) -> Result<Self, HttpError> {
-        let token = extract_session_token(&parts.headers).ok_or(StatusCode::UNAUTHORIZED)?;
+        let token = extract_session_token(&parts.headers).ok_or(HttpError::Unauthorized)?;
 
         let user = users::Entity::find_by_auth_session(&state.db, token)
-            .await
+            .await?
-            .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
+            .ok_or(HttpError::Unauthorized)?;
-            .ok_or(StatusCode::UNAUTHORIZED)?;
 
         Ok(AuthUser(user))
     }
@@ -254,12 +245,11 @@ async fn create_stream_key_handler(
     State(state): State<Arc<HttpServer>>,
     auth: AuthUser,
     Json(payload): Json<CreateStreamKeyBody>,
-) -> impl IntoResponse {
+) -> Result<impl IntoResponse, HttpError> {
     let uuid = uuid::Uuid::new_v4();
     let value = format!("stream-key-{uuid}");
     let key_amount = stream_key::Entity::find_by_user(&state.db, auth.0.id)
-        .await
+        .await?
-        .unwrap()
         .len();
     if payload.label.is_empty() || payload.label.len() > MAX_LABEL_LEN {
         warn!(
@@ -268,7 +258,7 @@ async fn create_stream_key_handler(
             max = MAX_LABEL_LEN,
             "stream key creation rejected: label length invalid"
         );
-        return StatusCode::UNPROCESSABLE_ENTITY;
+        return Err(HttpError::Unprocessable("label length invalid".into()));
     }
     if key_amount >= auth.0.stream_key_limit.try_into().unwrap() {
         warn!(
@@ -276,16 +266,12 @@ async fn create_stream_key_handler(
             limit = auth.0.stream_key_limit,
             "stream key limit reached"
         );
-        return StatusCode::NOT_ACCEPTABLE;
+        return Err(HttpError::NotAcceptable("stream key limit reached".into()));
     }
-    let key = stream_key::Entity::create(&state.db, auth.0.id, value, payload.label, false).await;
+    let key = stream_key::Entity::create(&state.db, auth.0.id, value, payload.label, false).await?;
 
-    if let Ok(ref k) = key {
+    info!(user_id = auth.0.id, stream_key_id = key.id, label = %key.label, "stream key created");
-        info!(user_id = auth.0.id, stream_key_id = k.id, label = %k.label, "stream key created");
+    Ok(StatusCode::CREATED)
-        StatusCode::CREATED
-    } else {
-        StatusCode::INTERNAL_SERVER_ERROR
-    }
 }
 
 struct StreamKeys {
@@ -301,15 +287,9 @@ struct StreamKey {
 async fn get_all_stream_keys(
     State(state): State<Arc<HttpServer>>,
     auth: AuthUser,
-) -> impl IntoResponse {
+) -> Result<impl IntoResponse, HttpError> {
-    // let user_keys = stream_key::Entity::find_by_user(&state.db, auth.0.id)
+    let keys = stream_key::Entity::find_by_user(&state.db, auth.0.id).await?;
-    //     .await
+    Ok(Json(keys))
-    //     .unwrap();
-
-    match stream_key::Entity::find_by_user(&state.db, auth.0.id).await {
-        Ok(keys) => Json(keys).into_response(),
-        Err(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(),
-    }
 }
 
 #[derive(Deserialize)]
@@ -326,41 +306,33 @@ struct LoginResponse {
 async fn login_handler(
     State(state): State<Arc<HttpServer>>,
     Json(payload): Json<LoginForm>,
-) -> impl IntoResponse {
+) -> Result<impl IntoResponse, HttpError> {
-    if let Ok(x) = users::Entity::find_by_username(&state.db, payload.username.clone()).await {
+    let user = users::Entity::find_by_username(&state.db, payload.username.clone())
-        if let Some(x) = x {
+        .await?
-            let pass = verify_password(&payload.password, &x.hashed_password);
+        .ok_or_else(|| {
-            if !pass {
-                warn!(username = %payload.username, "login failed: wrong password");
-                let mut meow = Response::new("".to_string());
-                *meow.status_mut() = StatusCode::UNAUTHORIZED;
-                return meow;
-            };
-            info!(user_id = x.id, username = %x.username, "login successful");
-            let auth = auth_session::Entity::create(&state.db, x.id).await.unwrap();
-            let token = auth.value;
-            let mut meow = Response::new("".to_string());
-            // TODO: Replace with axum cookie jar https://docs.rs/axum-extra/latest/axum_extra/extract/cookie/struct.CookieJar.html
-            meow.headers_mut().insert(
-                SET_COOKIE,
-                format!("session={token}; SameSite=Strict; Path=/; Max-Age=2592000")
-                    .parse()
-                    .unwrap(),
-            );
-            *meow.status_mut() = StatusCode::OK;
-            return meow;
-        } else {
             warn!(username = %payload.username, "login failed: user not found");
-            let mut meow = Response::new("".to_string());
+            HttpError::Unauthorized
-            *meow.status_mut() = StatusCode::UNAUTHORIZED;
+        })?;
-            return meow;
+
-        }
+    if !verify_password(&payload.password, &user.hashed_password) {
-    } else {
+        warn!(username = %payload.username, "login failed: wrong password");
-        warn!(username = %payload.username, "login failed: DB error");
+        return Err(HttpError::Unauthorized);
-        let mut meow = Response::new("".to_string());
-        *meow.status_mut() = StatusCode::UNAUTHORIZED;
-        return meow;
     }
+
+    info!(user_id = user.id, username = %user.username, "login successful");
+    let auth = auth_session::Entity::create(&state.db, user.id).await?;
+    let token = auth.value;
+
+    let mut meow = Response::new("".to_string());
+    // TODO: Replace with axum cookie jar https://docs.rs/axum-extra/latest/axum_extra/extract/cookie/struct.CookieJar.html
+    meow.headers_mut().insert(
+        SET_COOKIE,
+        format!("session={token}; SameSite=Strict; Path=/; Max-Age=2592000")
+            .parse()
+            .unwrap(),
+    );
+    *meow.status_mut() = StatusCode::OK;
+    Ok(meow)
 }
 
 #[derive(Deserialize)]
@@ -373,34 +345,33 @@ struct CreateUserForm {
 async fn create_user_handler(
     State(state): State<Arc<HttpServer>>,
     Json(payload): Json<CreateUserForm>,
-) -> (HeaderMap, StatusCode) {
+) -> Result<(HeaderMap, StatusCode), HttpError> {
     if state.config.signup_code.is_empty() || payload.ref_token != state.config.signup_code {
         warn!(username = %payload.username, "signup rejected: invalid signup code");
-        return (HeaderMap::new(), StatusCode::UNAUTHORIZED);
+        return Err(HttpError::Unauthorized);
     }
     if payload.username.is_empty() || payload.username.len() > MAX_USERNAME_LEN {
         warn!(username = %payload.username, max = MAX_USERNAME_LEN, "signup rejected: username length invalid");
-        return (HeaderMap::new(), StatusCode::UNPROCESSABLE_ENTITY);
+        return Err(HttpError::Unprocessable("username length invalid".into()));
     }
 
-    let meow = users::Entity::create(
+    let user = users::Entity::create(
         &state.db,
         payload.username.clone(),
-        hash_password(&payload.password).unwrap(),
+        hash_password(&payload.password).map_err(|_| HttpError::Hash)?,
     )
-    .await;
+    .await
+    .map_err(|e| {
+        // A create failure here is almost always a username unique-constraint
+        // violation, so surface it as a conflict rather than a 500.
+        warn!(username = %payload.username, error = %e, "user creation failed (likely username conflict)");
+        HttpError::Conflict
+    })?;
 
-    // Create session
+    info!(user_id = user.id, username = %user.username, "user created");
-    let session: auth_session::Model = if let Ok(ref user) = meow {
+    let session = auth_session::Entity::create(&state.db, user.id).await?;
-        info!(user_id = user.id, username = %user.username, "user created");
-        auth_session::Entity::create(&state.db, user.id)
-            .await
-            .unwrap()
-    } else {
-        warn!(username = %payload.username, "user creation failed (likely username conflict)");
-        return (HeaderMap::new(), StatusCode::CONFLICT);
-    };
     let token = session.value;
+
     let mut headers = HeaderMap::new();
     // TODO: Replace with axum cookie jar https://docs.rs/axum-extra/latest/axum_extra/extract/cookie/struct.CookieJar.html
     headers.insert(
@@ -409,14 +380,14 @@ async fn create_user_handler(
             .parse()
             .unwrap(),
     );
-    (headers, StatusCode::OK)
+    Ok((headers, StatusCode::OK))
 }
 
 async fn stream_handler(
     State(state): State<Arc<HttpServer>>,
     Path(slug): Path<String>,
     body: String,
-) -> impl IntoResponse {
+) -> Result<impl IntoResponse, HttpError> {
     let request_id_clone = state.request_count.fetch_add(1, Ordering::Relaxed);
 
     let stream_key_id = {
@@ -427,47 +398,42 @@ async fn stream_handler(
             .map(|e| *e.key())
     };
 
-    let stream_key_id = if let Some(id) = stream_key_id {
+    let stream_key_id = stream_key_id.ok_or_else(|| {
-        id
-    } else {
         warn!(slug = %slug, "WHEP request for unknown or inactive stream");
-        return Response::builder()
+        HttpError::NotFound
-            .status(StatusCode::NOT_FOUND)
+    })?;
-            .header("content-type", "application/text")
-            .body("".to_string())
-            .unwrap();
-    };
 
     info!(request_id = request_id_clone, slug = %slug, stream_key_id, "WHEP offer received");
     let mut accept_rx = state.accept_rx.activate_cloned();
-    let _ = state
+    // The webrtc worker owning the receiver died if this fails.
+    state
         .offer_tx
         .send((request_id_clone, stream_key_id, body))
         .await
-        .unwrap();
+        .map_err(|_| HttpError::Internal)?;
     debug!(
         request_id = request_id_clone,
         "offer sent, waiting for answer"
     );
 
-    let mut reply_body = String::new();
+    let reply_body = String::new();
     while let Ok(answer) = accept_rx.recv().await {
         debug!(request_id = request_id_clone, "received answer candidate");
         if let Some(reply) = answer.1 {
             if answer.0 == request_id_clone {
-                return Response::builder()
+                return Ok(Response::builder()
                     .status(StatusCode::CREATED)
                     .header("content-type", "application/sdp")
                     .body(reply)
-                    .unwrap();
+                    .unwrap());
             } else {
                 continue;
             }
         } else if answer.0 == request_id_clone {
-            return Response::builder()
+            return Ok(Response::builder()
                 .status(StatusCode::UNSUPPORTED_MEDIA_TYPE)
                 .body(String::new())
-                .unwrap();
+                .unwrap());
         }
     }
     info!(
@@ -475,11 +441,11 @@ async fn stream_handler(
         "answer channel closed without a match"
     );
 
-    Response::builder()
+    Ok(Response::builder()
         .status(StatusCode::CREATED)
         .header("content-type", "application/sdp")
         .body(reply_body)
-        .unwrap()
+        .unwrap())
 }
 
 async fn meow_handler() -> &'static str {

crates/server/src/http_error.rs

@@ -0,0 +1,59 @@
+use axum::{
+    body::Body,
+    http::{Response, StatusCode},
+    response::IntoResponse,
+};
+use thiserror::Error;
+
+#[derive(Error, Debug)]
+pub enum HttpError {
+    #[error("db shit itself: {0}")]
+    DbErr(#[from] sea_orm::DbErr),
+    #[error("hashing failed")]
+    Hash,
+    #[error("resource not found")]
+    NotFound,
+    #[error("unauthorized")]
+    Unauthorized,
+    #[error("forbidden")]
+    Forbidden,
+    #[error("conflict")]
+    Conflict,
+    #[error("bad request: {0}")]
+    BadRequest(String),
+    #[error("unprocessable entity: {0}")]
+    Unprocessable(String),
+    #[error("not acceptable: {0}")]
+    NotAcceptable(String),
+    #[error("internal error")]
+    Internal,
+}
+
+impl HttpError {
+    fn status(&self) -> StatusCode {
+        match self {
+            Self::DbErr(_) | Self::Hash | Self::Internal => StatusCode::INTERNAL_SERVER_ERROR,
+            Self::NotFound => StatusCode::NOT_FOUND,
+            Self::Unauthorized => StatusCode::UNAUTHORIZED,
+            Self::Forbidden => StatusCode::FORBIDDEN,
+            Self::Conflict => StatusCode::CONFLICT,
+            Self::BadRequest(_) => StatusCode::BAD_REQUEST,
+            Self::Unprocessable(_) => StatusCode::UNPROCESSABLE_ENTITY,
+            Self::NotAcceptable(_) => StatusCode::NOT_ACCEPTABLE,
+        }
+    }
+}
+
+impl IntoResponse for HttpError {
+    fn into_response(self) -> Response<Body> {
+        let status = self.status();
+        // Only surface a message body for client (4xx) errors; keep an empty
+        // body for 5xx so internal details aren't leaked.
+        let body = if status.is_client_error() {
+            self.to_string()
+        } else {
+            String::new()
+        };
+        (status, body).into_response()
+    }
+}

crates/server/src/main.rs

@@ -35,6 +35,7 @@ mod audio;
 mod codec;
 mod hash;
 mod http;
+mod http_error;
 mod rtmp;
 mod webrtc;
 mod webrtc_ingest;