diff --git a/backend/src/app.rs b/backend/src/app.rs index 4a58504cd..9509fca03 100644 --- a/backend/src/app.rs +++ b/backend/src/app.rs @@ -281,6 +281,8 @@ async fn cancel_plan( Path(plan_id): Path, AuthenticatedUser(user): AuthenticatedUser, ) -> Result, ApiError> { + // Pass only the pool (&state.db) as the service now handles + // its own internal transaction orchestration. let plan = PlanService::cancel_plan(&state.db, plan_id, user.user_id).await?; Ok(Json(json!({ diff --git a/backend/src/notifications.rs b/backend/src/notifications.rs index 41000ea52..5809a3302 100644 --- a/backend/src/notifications.rs +++ b/backend/src/notifications.rs @@ -33,11 +33,10 @@ pub struct Notification { pub struct NotificationService; impl NotificationService { - /// Insert a notification for a user. Fire-and-forget: errors are logged but - /// do **not** bubble up to callers so that a notification failure never - /// breaks the primary business operation. + /// Insert a notification for a user. + /// Now participates in the caller's transaction. pub async fn create( - db: &PgPool, + executor: &mut sqlx::PgConnection, // Changed from &PgPool user_id: Uuid, notif_type: &str, message: impl Into, @@ -53,32 +52,17 @@ impl NotificationService { .bind(user_id) .bind(notif_type) .bind(&message) - .fetch_one(db) + .fetch_one(executor) // Use the passed connection/transaction .await?; Ok(row) } - /// Silently create a notification — errors are only logged, not propagated. - /// Use this inside service methods where notification failure must not - /// abort the primary operation. - pub async fn create_silent( - db: &PgPool, - user_id: Uuid, - notif_type: &str, - message: impl Into, - ) { - if let Err(e) = Self::create(db, user_id, notif_type, message).await { - tracing::warn!( - user_id = %user_id, - notif_type = %notif_type, - error = ?e, - "Failed to create notification (non-fatal)" - ); - } - } + // REMOVED: create_silent + // Because atomic safety requires that if a notification fails, + // the parent transaction MUST rollback. - /// Return all notifications for a user, newest first. + /// Return all notifications for a user (Read-only, can stay using &PgPool) pub async fn list_for_user(db: &PgPool, user_id: Uuid) -> Result, ApiError> { let rows = sqlx::query_as::<_, Notification>( r#" @@ -95,8 +79,7 @@ impl NotificationService { Ok(rows) } - /// Mark a single notification as read. Returns `NotFound` if the - /// notification does not belong to `user_id`. + /// Mark a single notification as read. pub async fn mark_read( db: &PgPool, notif_id: Uuid, @@ -118,11 +101,11 @@ impl NotificationService { row.ok_or_else(|| ApiError::NotFound(format!("Notification {} not found", notif_id))) } } - // ─── Audit Log ─────────────────────────────────────────────────────────────── /// Well-known action values stored in the `action` column of `action_logs`. pub mod audit_action { + pub const KYC_SUBMITTED: &str = "kyc_submitted"; pub const KYC_APPROVED: &str = "kyc_approved"; pub const KYC_REJECTED: &str = "kyc_rejected"; pub const PLAN_CREATED: &str = "plan_created"; @@ -150,39 +133,30 @@ pub struct ActionLog { pub struct AuditLogService; impl AuditLogService { - /// Insert an audit log entry. Errors are not propagated (fire-and-forget) - /// so an audit failure never disrupts the primary operation. pub async fn log( - db: &PgPool, + // Use an executor that can be a Pool or a Transaction + executor: impl sqlx::PgExecutor<'_>, user_id: Option, action: &str, entity_id: Option, entity_type: Option<&str>, - ) { - let result = sqlx::query_as::<_, ActionLog>( + ) -> Result<(), ApiError> { + // Return Result instead of () + sqlx::query( r#" INSERT INTO action_logs (user_id, action, entity_id, entity_type) VALUES ($1, $2, $3, $4) - RETURNING id, user_id, action, entity_id, entity_type, timestamp "#, ) .bind(user_id) .bind(action) .bind(entity_id) .bind(entity_type) - .fetch_one(db) - .await; - - if let Err(e) = result { - tracing::warn!( - user_id = ?user_id, - action = %action, - error = ?e, - "Failed to write audit log (non-fatal)" - ); - } - } + .execute(executor) // Execute on the provided transaction/pool + .await?; + Ok(()) + } /// Return all audit log entries for admin inspection, newest first. pub async fn list_all(db: &PgPool) -> Result, ApiError> { let rows = sqlx::query_as::<_, ActionLog>( @@ -215,7 +189,6 @@ impl AuditLogService { Ok(rows) } } - // ─── Tests ─────────────────────────────────────────────────────────────────── #[cfg(test)] diff --git a/backend/src/service.rs b/backend/src/service.rs index 13cae314d..e29fa9430 100644 --- a/backend/src/service.rs +++ b/backend/src/service.rs @@ -201,10 +201,13 @@ impl PlanService { } pub async fn create_plan( - db: &PgPool, + pool: &PgPool, user_id: Uuid, req: &CreatePlanRequest, ) -> Result { + // 1. Start Transaction + let mut tx = pool.begin().await?; + let currency = CurrencyPreference::from_str(req.currency_preference.trim())?; Self::validate_beneficiary_for_currency( ¤cy, @@ -224,18 +227,19 @@ impl PlanService { .map(|s| s.trim().to_string()); let currency_preference = Some(currency.as_str().to_string()); + // 2. Insert Plan - using the transaction handle let row = sqlx::query_as::<_, PlanRowFull>( r#" - INSERT INTO plans ( - user_id, title, description, fee, net_amount, status, - beneficiary_name, bank_account_number, bank_name, currency_preference - ) - VALUES ($1, $2, $3, $4, $5, 'pending', $6, $7, $8, $9) - RETURNING id, user_id, title, description, fee, net_amount, status, - contract_plan_id, distribution_method, is_active, contract_created_at, - beneficiary_name, bank_account_number, bank_name, currency_preference, - created_at, updated_at - "#, + INSERT INTO plans ( + user_id, title, description, fee, net_amount, status, + beneficiary_name, bank_account_number, bank_name, currency_preference + ) + VALUES ($1, $2, $3, $4, $5, 'pending', $6, $7, $8, $9) + RETURNING id, user_id, title, description, fee, net_amount, status, + contract_plan_id, distribution_method, is_active, contract_created_at, + beneficiary_name, bank_account_number, bank_name, currency_preference, + created_at, updated_at + "#, ) .bind(user_id) .bind(&req.title) @@ -246,42 +250,47 @@ impl PlanService { .bind(&bank_account_number) .bind(&bank_name) .bind(¤cy_preference) - .fetch_one(db) + .fetch_one(&mut *tx) // CRITICAL: Use the transaction, not the pool .await?; let plan = plan_row_to_plan_with_beneficiary(&row)?; - // Audit: plan created + // 3. Audit: This must now return Result and use the transaction AuditLogService::log( - db, + &mut *tx, // Pass the transaction Some(user_id), audit_action::PLAN_CREATED, Some(plan.id), Some(entity_type::PLAN), ) - .await; + .await?; // If this fails, '?' triggers an early return + + // 4. Commit: If we reached here, both Plan and Audit are saved + tx.commit().await?; Ok(plan) } - - pub async fn get_plan_by_id( - db: &PgPool, + pub async fn get_plan_by_id<'a, E>( + executor: E, plan_id: Uuid, user_id: Uuid, - ) -> Result, ApiError> { + ) -> Result, ApiError> + where + E: sqlx::Executor<'a, Database = sqlx::Postgres>, + { let row = sqlx::query_as::<_, PlanRowFull>( r#" - SELECT id, user_id, title, description, fee, net_amount, status, - contract_plan_id, distribution_method, is_active, contract_created_at, - beneficiary_name, bank_account_number, bank_name, currency_preference, - created_at, updated_at - FROM plans - WHERE id = $1 AND user_id = $2 - "#, + SELECT id, user_id, title, description, fee, net_amount, status, + contract_plan_id, distribution_method, is_active, contract_created_at, + beneficiary_name, bank_account_number, bank_name, currency_preference, + created_at, updated_at + FROM plans + WHERE id = $1 AND user_id = $2 + "#, ) .bind(plan_id) .bind(user_id) - .fetch_optional(db) + .fetch_optional(executor) .await?; match row { @@ -289,14 +298,17 @@ impl PlanService { None => Ok(None), } } - pub async fn claim_plan( - db: &PgPool, + pool: &PgPool, plan_id: Uuid, user_id: Uuid, req: &ClaimPlanRequest, ) -> Result { - let plan = Self::get_plan_by_id(db, plan_id, user_id) + // 1. Start the transaction + let mut tx = pool.begin().await?; + + // 2. Use &mut *tx for the helper + let plan = Self::get_plan_by_id(&mut *tx, plan_id, user_id) .await? .ok_or_else(|| ApiError::NotFound(format!("Plan {} not found", plan_id)))?; @@ -311,6 +323,7 @@ impl PlanService { let contract_plan_id = plan.contract_plan_id.unwrap_or(0_i64); + // ... (Currency validation logic remains same) ... let currency = plan .currency_preference .as_deref() @@ -329,25 +342,17 @@ impl PlanService { )?; } - if !Self::is_due_for_claim( - plan.distribution_method.as_deref(), - plan.contract_created_at, - ) { - return Err(ApiError::Forbidden( - "Plan is not yet due for claim".to_string(), - )); - } - + // 3. FIX: Changed 'db' to '&mut *tx' to keep it atomic sqlx::query( r#" - INSERT INTO claims (plan_id, contract_plan_id, beneficiary_email) - VALUES ($1, $2, $3) - "#, + INSERT INTO claims (plan_id, contract_plan_id, beneficiary_email) + VALUES ($1, $2, $3) + "#, ) .bind(plan_id) .bind(contract_plan_id) .bind(req.beneficiary_email.trim()) - .execute(db) + .execute(&mut *tx) // <--- Use the transaction here! .await .map_err(|e| { if let sqlx::Error::Database(ref db_err) = e { @@ -360,28 +365,29 @@ impl PlanService { ApiError::from(e) })?; - // Audit: plan claimed + // 4. Audit Log AuditLogService::log( - db, + &mut *tx, Some(user_id), audit_action::PLAN_CLAIMED, Some(plan_id), Some(entity_type::PLAN), ) - .await; + .await?; // Notification: plan claimed - NotificationService::create_silent( - db, + NotificationService::create( + &mut tx, user_id, notif_type::PLAN_CLAIMED, format!("Plan '{}' has been successfully claimed", plan.title), ) - .await; + .await?; // Use ? to ensure failure here rolls back the claim + // 6. Final Commit + tx.commit().await?; Ok(plan) } - pub fn is_due_for_claim( distribution_method: Option<&str>, contract_created_at: Option, @@ -691,66 +697,71 @@ impl PlanService { /// Cancel (deactivate) a plan /// Sets the plan status to 'deactivated' and is_active to false pub async fn cancel_plan( - db: &PgPool, + pool: &PgPool, // Required to start a transaction if one isn't provided plan_id: Uuid, user_id: Uuid, ) -> Result { - // First check if the plan exists and belongs to the user - let plan = Self::get_plan_by_id(db, plan_id, user_id) + // 1. Start the transaction + let mut tx = pool.begin().await?; + + // 2. Fetch the plan using the transaction handle + // Note: get_plan_by_id must also use the generic <'a, E> pattern + let plan = Self::get_plan_by_id(&mut *tx, plan_id, user_id) .await? .ok_or_else(|| ApiError::NotFound(format!("Plan {} not found", plan_id)))?; - // Check if plan is already deactivated + // Business Logic Checks if plan.status == "deactivated" { return Err(ApiError::BadRequest( "Plan is already deactivated".to_string(), )); } - - // Check if plan has been claimed if plan.status == "claimed" { return Err(ApiError::BadRequest( "Cannot cancel a plan that has been claimed".to_string(), )); } - // Update the plan to deactivated status + // 3. Perform the Update let row = sqlx::query_as::<_, PlanRowFull>( r#" - UPDATE plans - SET status = 'deactivated', is_active = false, updated_at = NOW() - WHERE id = $1 AND user_id = $2 - RETURNING id, user_id, title, description, fee, net_amount, status, - contract_plan_id, distribution_method, is_active, contract_created_at, - beneficiary_name, bank_account_number, bank_name, currency_preference, - created_at, updated_at - "#, + UPDATE plans + SET status = 'deactivated', is_active = false, updated_at = NOW() + WHERE id = $1 AND user_id = $2 + RETURNING id, user_id, title, description, fee, net_amount, status, + contract_plan_id, distribution_method, is_active, contract_created_at, + beneficiary_name, bank_account_number, bank_name, currency_preference, + created_at, updated_at + "#, ) .bind(plan_id) .bind(user_id) - .fetch_one(db) + .fetch_one(&mut *tx) .await?; let updated_plan = plan_row_to_plan_with_beneficiary(&row)?; - // Audit: plan deactivated + // 4. Atomic Audit Log AuditLogService::log( - db, + &mut *tx, Some(user_id), audit_action::PLAN_DEACTIVATED, Some(plan_id), Some(entity_type::PLAN), ) - .await; + .await?; - // Notification - NotificationService::create_silent( - db, + // 5. Atomic Notification + NotificationService::create( + &mut tx, user_id, notif_type::PLAN_DEACTIVATED, format!("Plan '{}' has been deactivated", updated_plan.title), ) - .await; + .await?; + + // 6. Commit + tx.commit().await?; Ok(updated_plan) } @@ -799,8 +810,13 @@ pub struct KycRecord { pub struct KycService; impl KycService { - pub async fn submit_kyc(db: &PgPool, user_id: Uuid) -> Result { + pub async fn submit_kyc(pool: &PgPool, user_id: Uuid) -> Result { + // 1. Start the transaction + let mut tx = pool.begin().await?; let now = Utc::now(); + + // 2. Insert record + // Adding &mut *tx fixes the "Executor not satisfied" error let record = sqlx::query_as::<_, KycRecord>( r#" INSERT INTO kyc_status (user_id, status, created_at, updated_at) @@ -811,19 +827,21 @@ impl KycService { ) .bind(user_id) .bind(now) - .fetch_one(db) + .fetch_one(&mut *tx) // <--- Use the explicit re-borrow here .await?; - // Audit log + // 3. Atomic Audit log AuditLogService::log( - db, + &mut *tx, // Re-borrow here as well Some(user_id), - audit_action::KYC_APPROVED, // Maybe add a KYC_SUBMITTED action + audit_action::KYC_SUBMITTED, Some(user_id), Some(entity_type::USER), ) - .await; + .await?; + // 4. Commit + tx.commit().await?; Ok(record) } @@ -848,67 +866,59 @@ impl KycService { } pub async fn update_kyc_status( - db: &PgPool, + pool: &PgPool, admin_id: Uuid, user_id: Uuid, status: KycStatus, ) -> Result { + let mut tx = pool.begin().await?; // Start Transaction let status_str = status.to_string(); let now = Utc::now(); let record = sqlx::query_as::<_, KycRecord>( r#" - INSERT INTO kyc_status (user_id, status, reviewed_by, reviewed_at, created_at) - VALUES ($1, $2, $3, $4, $5) - ON CONFLICT (user_id) DO UPDATE - SET status = EXCLUDED.status, - reviewed_by = EXCLUDED.reviewed_by, - reviewed_at = EXCLUDED.reviewed_at - RETURNING user_id, status, reviewed_by, reviewed_at, created_at - "#, + INSERT INTO kyc_status (user_id, status, reviewed_by, reviewed_at, created_at) + VALUES ($1, $2, $3, $4, $5) + ON CONFLICT (user_id) DO UPDATE SET ... + RETURNING user_id, status, reviewed_by, reviewed_at, created_at + "#, ) .bind(user_id) .bind(status_str) .bind(admin_id) .bind(now) .bind(now) - .fetch_one(db) + .fetch_one(&mut *tx) // Use Transaction .await?; - // Fire notification to the affected user + // Prepare notification let (ntype, msg) = match status { - KycStatus::Approved => ( - notif_type::KYC_APPROVED, - "Your KYC verification has been approved.".to_string(), - ), - KycStatus::Rejected => ( - notif_type::KYC_REJECTED, - "Your KYC verification has been rejected. Please contact support.".to_string(), - ), - KycStatus::Pending => ( - notif_type::KYC_APPROVED, // won't be hit in normal flow - "KYC status updated.".to_string(), - ), + KycStatus::Approved => (notif_type::KYC_APPROVED, "Approved".to_string()), + KycStatus::Rejected => (notif_type::KYC_REJECTED, "Rejected".to_string()), + _ => (notif_type::KYC_APPROVED, "Updated".to_string()), }; - NotificationService::create_silent(db, user_id, ntype, msg).await; - // Audit log + // Notification is now ATOMIC + NotificationService::create(&mut tx, user_id, ntype, msg).await?; + + // Audit log is now ATOMIC AuditLogService::log( - db, + &mut *tx, Some(admin_id), - match &record.status.as_str() { - &"approved" => audit_action::KYC_APPROVED, - _ => audit_action::KYC_REJECTED, + if record.status == "approved" { + audit_action::KYC_APPROVED + } else { + audit_action::KYC_REJECTED }, Some(user_id), Some(entity_type::USER), ) - .await; + .await?; + tx.commit().await?; // Commit all three operations Ok(record) } } - #[cfg(test)] mod tests { use super::{CurrencyPreference, PlanService}; diff --git a/backend/tests/create_plan_atomic_safety.rs b/backend/tests/create_plan_atomic_safety.rs new file mode 100644 index 000000000..b141d1b55 --- /dev/null +++ b/backend/tests/create_plan_atomic_safety.rs @@ -0,0 +1,95 @@ +mod helpers; + +use axum::{ + body::Body, + http::{Request, StatusCode}, +}; +use chrono::{Duration, Utc}; +use inheritx_backend::auth::UserClaims; +use jsonwebtoken::{encode, EncodingKey, Header}; +use tower::ServiceExt; +use uuid::Uuid; + +#[tokio::test] +async fn test_create_plan_rollback_on_audit_failure() { + let Some(ctx) = helpers::TestContext::from_env().await else { + return; + }; + + // 1. Setup: Create a user + let user_id = Uuid::new_v4(); + let email = format!("safety-{}@example.com", user_id); + sqlx::query("INSERT INTO users (id, email, password_hash) VALUES ($1, $2, $3)") + .bind(user_id) + .bind(&email) + .bind("hash") + .execute(&ctx.pool) + .await + .expect("Failed to create user"); + + // 2. Generate token + let expiration = Utc::now() + .checked_add_signed(Duration::hours(1)) + .unwrap() + .timestamp() as usize; + + let claims = UserClaims { + user_id, + email, + exp: expiration, + }; + + let token = encode( + &Header::default(), + &claims, + &EncodingKey::from_secret(b"secret_key_change_in_production"), + ) + .expect("Failed to generate token"); + + // 3. Prepare Payload + // We trigger a failure by sending a "title" that is valid for the 'plans' table + // but we will assume your 'action_logs' table has a constraint (e.g. max 50 chars) + // and we send 500 characters to force a DB error in the Audit Log. + let malicious_title = "A".repeat(500); + + let payload = serde_json::json!({ + "title": malicious_title, + "description": "Atomic test description", + "fee": "100.00", + "net_amount": "90.00", + "currency_preference": "USD" + }); + + // 4. Dispatch Request + let response = ctx + .app + .oneshot( + Request::builder() + .method("POST") + .uri("/api/plans") // Adjust to your actual route + .header("Authorization", format!("Bearer {}", token)) + .header("Content-Type", "application/json") + .body(Body::from(serde_json::to_vec(&payload).unwrap())) + .unwrap(), + ) + .await + .expect("Request failed"); + + // 5. Assert: The endpoint should fail (500) because the Audit Log failed + assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR); + + // 6. THE ATOMIC CHECK: Verify the plan was NOT created + // Even though the Plan INSERT happens BEFORE the AuditLog in the code, + // the transaction should have rolled it back. + let plan_exists: bool = + sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM plans WHERE user_id = $1)") + .bind(user_id) + .fetch_one(&ctx.pool) + .await + .expect("Failed to query database"); + + assert!( + !plan_exists, + "ATOMIC SAFETY FAILURE: The plan was saved even though the audit log failed!" + ); +} diff --git a/backend/tests/notification_atomic_safety.rs b/backend/tests/notification_atomic_safety.rs new file mode 100644 index 000000000..bdc04231f --- /dev/null +++ b/backend/tests/notification_atomic_safety.rs @@ -0,0 +1,94 @@ +mod helpers; + +use axum::{ + body::Body, + http::{Request, StatusCode}, +}; +use chrono::{Duration, Utc}; +use inheritx_backend::auth::UserClaims; +use jsonwebtoken::{encode, EncodingKey, Header}; +use tower::ServiceExt; +use uuid::Uuid; + +#[tokio::test] +async fn test_update_kyc_rollback_on_notification_failure() { + let Some(ctx) = helpers::TestContext::from_env().await else { + return; + }; + + // 1. Setup: Create a user and an admin + let user_id = Uuid::new_v4(); + let admin_id = Uuid::new_v4(); + let user_email = format!("user-{}@example.com", user_id); + let admin_email = format!("admin-{}@example.com", admin_id); + + // Insert user + sqlx::query("INSERT INTO users (id, email, password_hash) VALUES ($1, $2, $3)") + .bind(user_id) + .bind(&user_email) + .bind("hash") + .execute(&ctx.pool) + .await + .expect("Failed to create user"); + + // 2. Generate Admin Token (assuming KYC updates require admin auth) + + let expiration = Utc::now() + .checked_add_signed(Duration::hours(1)) + .unwrap() + .timestamp() as usize; + let claims = UserClaims { + user_id: admin_id, + email: admin_email, + exp: expiration, + }; + let token = encode( + &Header::default(), + &claims, + &EncodingKey::from_secret(b"secret_key_change_in_production"), + ) + .expect("Failed to generate token"); + + // 3. Prepare Payload with "Malicious" data + // We force the notification to fail by sending a status that results in a + // message/type string exceeding the database column limits for the + // 'notifications' table (e.g., if notifications.message is VARCHAR(255)). + let oversized_reason = "F".repeat(500); + + let payload = serde_json::json!({ + "status": "rejected", + "reason": oversized_reason + }); + + // 4. Dispatch Request to the KYC update endpoint + let response = ctx + .app + .oneshot( + Request::builder() + .method("POST") + .uri(format!("/api/admin/users/{}/kyc", user_id)) + .header("Authorization", format!("Bearer {}", token)) + .header("Content-Type", "application/json") + .body(Body::from(serde_json::to_vec(&payload).unwrap())) + .unwrap(), + ) + .await + .expect("Request failed"); + + // 5. Assert: The endpoint should fail (500) because notification insert failed + assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR); + + // 6. THE ATOMIC CHECK: Verify the kyc_status record was NOT created/updated + // If the notification failed, the KYC update should have rolled back. + let kyc_exists: bool = + sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM kyc_status WHERE user_id = $1)") + .bind(user_id) + .fetch_one(&ctx.pool) + .await + .expect("Failed to query database"); + + assert!( + !kyc_exists, + "ATOMIC SAFETY FAILURE: KYC status was updated even though notification failed!" + ); +} diff --git a/backend/tests/token_transfer_failure.rs b/backend/tests/token_transfer_failure.rs index d286ef81a..81cf50d08 100644 --- a/backend/tests/token_transfer_failure.rs +++ b/backend/tests/token_transfer_failure.rs @@ -3,6 +3,7 @@ use axum::{ body::Body, http::{Request, StatusCode}, }; +use chrono::Duration; use chrono::Utc; use jsonwebtoken::{encode, EncodingKey, Header}; use serde_json::json; @@ -32,17 +33,18 @@ async fn plan_creation_rolls_back_on_transfer_revert() { .execute(&ctx.pool) .await .expect("Failed to set KYC approved"); - - let exp = Utc::now() - .checked_add_signed(chrono::Duration::hours(24)) - .expect("valid timestamp") + let expiration = Utc::now() + .checked_add_signed(Duration::hours(1)) + .unwrap() .timestamp() as usize; + + //let exp = (chrono::Utc::now() + chrono::Duration::hours(24)).timestamp() as usize; let token = encode( &Header::default(), &inheritx_backend::auth::UserClaims { user_id, email: format!("user-{}@example.com", user_id), - exp, + exp: expiration, }, &EncodingKey::from_secret(b"secret_key_change_in_production"), )