fix(backend): cleanups part 1

backend/backend_config.toml

@@ -26,9 +26,6 @@ jwt_secret = "0xdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeef
 # WebSocket RPC endpoint for the StorageHub node
 # Use ws:// for unencrypted or wss:// for TLS-encrypted connections
 rpc_url = "ws://localhost:9944"
-# URL for node to communicate with the backend (e.g for uploading a file)
-# e.g: if running an msp in a container: http://host.docker.internal:8080
-msp_callback_url = "http://localhost:8080"
 # Request timeout in seconds (optional, default: 30)
 # Increase this value for slower networks or heavy operations
 timeout_secs = 30
@@ -42,6 +39,17 @@ verify_tls = true
 # When true, uses mock RPC connections instead of real network calls
 # mock_mode = false
 
+# MSP-specific configuration
+[msp]
+# URL for node to communicate with the backend (e.g for uploading a file)
+# e.g: if running an msp in a container: http://host.docker.internal:8080
+callback_url = "http://localhost:8080"
+# Number of retry attempts for file upload operations (default: 3)
+# Increase for unreliable network conditions
+upload_retry_attempts = 3
+# Delay in seconds between file upload retry attempts (default: 1)
+upload_retry_delay_secs = 1
+
 # Database configuration
 [database]
 # PostgreSQL connection URL

backend/bin/src/main.rs

@@ -74,7 +74,7 @@ async fn main() -> Result<()> {
     );
     debug!(target: "main", database_url = %config.database.url, "Database configuration");
     debug!(target: "main", rpc_url = %config.storage_hub.rpc_url, "RPC configuration");
-    debug!(target: "main", msp_callback_url = %config.storage_hub.msp_callback_url, "MSP callback configuration");
+    debug!(target: "main", msp_callback_url = %config.msp.callback_url, "MSP callback configuration");
 
     let memory_storage = InMemoryStorage::new();
     let storage = Arc::new(BoxedStorageWrapper::new(memory_storage));
@@ -135,7 +135,7 @@ fn load_config() -> Result<Config> {
         config.storage_hub.rpc_url = rpc_url;
     }
     if let Some(msp_callback_url) = args.msp_callback_url {
-        config.storage_hub.msp_callback_url = msp_callback_url;
+        config.msp.callback_url = msp_callback_url;
     }
 
     Ok(config)

backend/lib/Cargo.toml

@@ -15,7 +15,7 @@ chrono = { workspace = true, features = ["serde"] }
 diesel = { workspace = true }
 diesel-async = { workspace = true }
 futures = { workspace = true }
-alloy-core = "0.8"
+alloy-core = { version = "0.8", features = ["serde"] }
 alloy-signer = "0.8"
 headers = { workspace = true }
 hex = { workspace = true }

backend/lib/src/api/handlers/auth.rs

@@ -93,15 +93,15 @@ mod tests {
 
         // Step 1: Get nonce challenge
         let nonce_request = NonceRequest {
-            address: address.clone(),
+            address,
             chain_id: 1,
         };
 
         let response = server.post(AUTH_NONCE_ENDPOINT).json(&nonce_request).await;
 
         assert_eq!(response.status_code(), StatusCode::OK);
         let nonce_response: NonceResponse = response.json();
-        assert!(nonce_response.message.contains(&address));
+        assert!(nonce_response.message.contains(&address.to_string()));
 
         // Step 2: Sign the message and login
         let signature = sign_message(&signing_key, &nonce_response.message);
@@ -163,17 +163,14 @@ mod tests {
         let app = mock_app().await;
         let server = TestServer::new(app).unwrap();
 
-        let invalid_request = NonceRequest {
-            address: "not_an_eth_address".to_string(),
-            chain_id: 1,
-        };
+        let invalid_json = serde_json::json!({
+            "address": "not_an_eth_address",
+            "chainId": 1
+        });
 
-        let response = server
-            .post(AUTH_NONCE_ENDPOINT)
-            .json(&invalid_request)
-            .await;
+        let response = server.post(AUTH_NONCE_ENDPOINT).json(&invalid_json).await;
 
-        assert_eq!(response.status_code(), StatusCode::BAD_REQUEST);
+        assert_eq!(response.status_code(), StatusCode::UNPROCESSABLE_ENTITY);
     }
 
     #[tokio::test]
@@ -299,7 +296,7 @@ mod tests {
         // Unfortunately we can't easily advance system time
         // so instead we create an "old" token
         let old_claims = JwtClaims {
-            address: MOCK_ADDRESS.to_string(),
+            address: MOCK_ADDRESS,
             iat: Utc::now().timestamp() - 10, // issued 10 seconds ago
             exp: Utc::now().timestamp() + 10, // expires in 10 seconds
         };
@@ -408,7 +405,7 @@ mod tests {
 
         // Login with first wallet
         let nonce_request1 = NonceRequest {
-            address: address1.clone(),
+            address: address1,
             chain_id: 1,
         };
 
@@ -428,7 +425,7 @@ mod tests {
 
         // Login with second wallet
         let nonce_request2 = NonceRequest {
-            address: address2.clone(),
+            address: address2,
             chain_id: 1,
         };
 
@@ -485,7 +482,7 @@ mod tests {
         // until the token is expired
         // so we create a token that's already expired
         let expired_claims = JwtClaims {
-            address: MOCK_ADDRESS.to_string(),
+            address: MOCK_ADDRESS,
             exp: Utc::now().timestamp() - 3600, // 1 hour ago
             iat: Utc::now().timestamp() - 7200, // 2 hours ago
         };

backend/lib/src/api/handlers/buckets.rs

@@ -9,18 +9,19 @@ use serde::Deserialize;
 use tracing::debug;
 
 use crate::{
-    error::Error, models::files::FileListResponse, services::auth::AuthenticatedUser,
-    services::Services,
+    api::handlers::pagination::Pagination, error::Error, models::files::FileListResponse,
+    services::auth::AuthenticatedUser, services::Services,
 };
 
 pub async fn list_buckets(
     State(services): State<Services>,
     AuthenticatedUser { address }: AuthenticatedUser,
+    Pagination { limit, offset }: Pagination,
 ) -> Result<impl IntoResponse, Error> {
     debug!(user = %address, "GET list buckets");
     let response = services
         .msp
-        .list_user_buckets(&address)
+        .list_user_buckets(&address, offset, limit)
         .await?
         .collect::<Vec<_>>();
     Ok(Json(response))
@@ -47,6 +48,7 @@ pub async fn get_files(
     AuthenticatedUser { address }: AuthenticatedUser,
     Path(bucket_id): Path<String>,
     Query(query): Query<FilesQuery>,
+    Pagination { limit, offset }: Pagination,
 ) -> Result<impl IntoResponse, Error> {
     let path = query.path.as_deref().unwrap_or("/");
     debug!(
@@ -55,14 +57,15 @@ pub async fn get_files(
         user = %address,
         "GET bucket files"
     );
+
     let file_tree = services
         .msp
-        .get_file_tree(&bucket_id, &address, path)
+        .get_file_tree(&bucket_id, &address, path, offset, limit)
         .await?;
 
     let response = FileListResponse {
         bucket_id: bucket_id.clone(),
-        files: vec![file_tree],
+        tree: file_tree,
     };
 
     Ok(Json(response))

backend/lib/src/api/handlers/files.rs

@@ -1,6 +1,4 @@
 //! This module contains the handlers for the file management endpoints
-//!
-//! TODO: move the rest of the endpoints as they are implemented
 
 use axum::{
     body::{Body, Bytes},
@@ -30,32 +28,29 @@ use crate::{
 pub async fn get_file_info(
     State(services): State<Services>,
     AuthenticatedUser { address }: AuthenticatedUser,
-    Path((bucket_id, file_key)): Path<(String, String)>,
+    Path((_bucket_id, file_key)): Path<(String, String)>,
 ) -> Result<impl IntoResponse, Error> {
     debug!(
-        bucket_id = %bucket_id,
         file_key = %file_key,
         user = %address,
         "GET file info"
     );
-    let response = services
-        .msp
-        .get_file_info(&bucket_id, &address, &file_key)
-        .await?;
+    let response = services.msp.get_file_info(&address, &file_key).await?;
     Ok(Json(response))
 }
 
-// Internal endpoint used by the MSP RPC to upload a file to the backend
-// This function streams the file chunks via a channel to the thread running download_by_key.
+/// Internal endpoint used by the MSP RPC to upload a file to the backend
+///
+/// This function streams the file chunks via a channel to the thread running download_by_key.
+// TODO(AUTH): Add MSP Node authentication
+// Currently this internal endpoint doesn't authenticate that
+// the client connecting to it is the MSP Node
 pub async fn internal_upload_by_key(
     State(services): State<Services>,
     Path((session_id, file_key)): Path<(String, String)>,
     body: Body,
 ) -> (StatusCode, impl IntoResponse) {
     debug!(file_key = %file_key, "PUT internal upload");
-    // TODO: re-add auth
-    // FIXME: make this only callable by the rpc itself
-    // let _auth = extract_bearer_token(&auth)?;
 
     // Validate file_key is a hex string
     let key = file_key.trim_start_matches("0x");
@@ -113,7 +108,7 @@ pub async fn download_by_key(
     Path(file_key): Path<String>,
 ) -> Result<impl IntoResponse, Error> {
     debug!(file_key = %file_key, user = %address, "GET download file");
-    // TODO(AUTH): verify that user has permissions to access this file
+
     // Validate file_key is a hex string
     let key = file_key.trim_start_matches("0x");
     if hex::decode(key).is_err() {
@@ -123,6 +118,9 @@ pub async fn download_by_key(
     // Check if file exists in MSP storage
     let file_metadata = services.msp.check_file_status(&file_key).await?;
 
+    // Verify user has access to the requested file
+    let file_info = services.msp.get_file_info(&address, &file_key).await?;
+
     // Generate a unique session ID for the download session
     let session_id = Uuid::now_v7().to_string();
 
@@ -138,13 +136,9 @@ pub async fn download_by_key(
         .add_session(&session_id, tx)
         .map_err(|e| Error::BadRequest(e.to_string()))?;
 
-    let file_key_clone = file_key.clone();
     tokio::spawn(async move {
         // We trigger the download process via RPC call
-        let _ = services
-            .msp
-            .get_file_from_key(&session_id, &file_key_clone)
-            .await;
+        _ = services.msp.get_file(&session_id, file_info).await;
     });
 
     // Extract filename from location or use file_key as fallback
@@ -182,16 +176,14 @@ pub async fn download_by_key(
 pub async fn upload_file(
     State(services): State<Services>,
     AuthenticatedUser { address }: AuthenticatedUser,
-    Path((bucket_id, file_key)): Path<(String, String)>,
+    Path((_bucket_id, file_key)): Path<(String, String)>,
     mut multipart: Multipart,
 ) -> Result<impl IntoResponse, Error> {
     debug!(
-        bucket_id = %bucket_id,
         file_key = %file_key,
         user = %address,
         "PUT upload file"
     );
-    // TODO(AUTH): verify that user has permissions to access this file
 
     // Pre-check with MSP whether this file key is expected before doing heavy processing
     let is_expected = services
@@ -252,25 +244,8 @@ pub async fn upload_file(
     // Process and upload the file using the MSP service
     let response = services
         .msp
-        .process_and_upload_file(&bucket_id, &file_key, file_data_stream, file_metadata)
+        .process_and_upload_file(&address, &file_key, file_data_stream, file_metadata)
         .await?;
 
     Ok((StatusCode::CREATED, Json(response)))
 }
-
-pub async fn distribute_file(
-    State(services): State<Services>,
-    AuthenticatedUser { address }: AuthenticatedUser,
-    Path((bucket_id, file_key)): Path<(String, String)>,
-) -> Result<impl IntoResponse, Error> {
-    debug!(
-        bucket_id = %bucket_id,
-        file_key = %file_key,
-        user = %address,
-        "POST distribute file"
-    );
-    // TODO(AUTH): verify that user has permissions to access this file
-
-    let response = services.msp.distribute_file(&bucket_id, &file_key).await?;
-    Ok(Json(response))
-}

backend/lib/src/api/handlers/mod.rs

@@ -10,6 +10,8 @@ pub mod auth;
 pub mod buckets;
 pub mod files;
 
+mod pagination;
+
 // ==================== MSP Info Handlers ====================
 
 pub async fn info(State(services): State<Services>) -> Result<impl IntoResponse, Error> {

backend/lib/src/api/handlers/pagination.rs

@@ -0,0 +1,61 @@
+use axum::{
+    extract::{rejection::QueryRejection, FromRequestParts, Query},
+    http::request::Parts,
+};
+use serde::Deserialize;
+
+use crate::constants::database::{DEFAULT_PAGE_LIMIT, MAX_PAGE_LIMIT};
+
+// NOTE: we use i64 because the db uses i64
+/// Resolved pagination parameters for the given endpoint
+///
+/// The query parameters used for the requests are the ones described in [`PaginationQuery`]
+///
+/// Values:
+/// * `limit`: the maximum amount of items to respond with
+/// * `offset`: the number of items to skip for this request
+#[derive(Debug)]
+pub struct Pagination {
+    pub limit: i64,
+    pub offset: i64,
+}
+
+impl From<PaginationQuery> for Pagination {
+    fn from(value: PaginationQuery) -> Self {
+        let limit = value.limit.unwrap_or(DEFAULT_PAGE_LIMIT).clamp(
+            0,
+            MAX_PAGE_LIMIT
+                .try_into()
+                .expect("MAX_PAGE_LIMIT to be representable as i64"),
+        );
+
+        let offset = limit * value.page.unwrap_or(0);
+
+        Self { offset, limit }
+    }
+}
+
+// NOTE: we use i64 because the db uses i64
+/// Pagination query parameters for the given endpoint
+///
+/// Parameters:
+/// * `limit`: the maximum amount of items to respond with (defaults to 0, capped at [`MAX_PAGE_LIMIT`])
+/// * `page`: the number of pages to skip for this request (defaults to 0), the number of elements in the page is equal to `limit`
+#[derive(Debug, Deserialize)]
+pub struct PaginationQuery {
+    page: Option<i64>,
+    limit: Option<i64>,
+}
+
+impl<S> FromRequestParts<S> for Pagination
+where
+    S: Send + Sync,
+{
+    type Rejection = QueryRejection;
+
+    async fn from_request_parts(parts: &mut Parts, state: &S) -> Result<Self, Self::Rejection> {
+        let query = Query::<PaginationQuery>::from_request_parts(parts, state).await?;
+
+        Ok(query.0.into())
+    }
+}

backend/lib/src/api/routes.rs

@@ -52,10 +52,6 @@ pub fn routes(services: Services) -> Router {
         )
         .merge(file_upload)
         .merge(internal_file_upload)
-        .route(
-            "/buckets/{bucket_id}/distribute/{file_key}",
-            post(handlers::files::distribute_file),
-        )
         .route(
             "/download/{file_key}",
             get(handlers::files::download_by_key),