diff --git a/api-server/src/handlers.rs b/api-server/src/handlers.rs index fa44e35..403ecee 100644 --- a/api-server/src/handlers.rs +++ b/api-server/src/handlers.rs @@ -7,6 +7,7 @@ use axum::{ use once_cell::sync::Lazy; use serde_json::Value; use std::collections::HashSet; +use std::sync::atomic::{AtomicU64, Ordering}; use tokio::time::{Duration, Instant}; use tracing::instrument; use crate::cache; @@ -17,6 +18,24 @@ use crate::webhook; // #523: Per-handler idempotency store for batch swap operations. static BATCH_SWAP_IDEMPOTENCY: Lazy = Lazy::new(create_store); +// #520: Mirrors the contract's MAX_BATCH_SIZE cap on a single batch. +const MAX_BATCH_SIZE: usize = 50; + +// #469: Mirrors the contract's ~7-day swap expiry (ledger timestamp + 604800). +const SWAP_EXPIRY_SECONDS: u64 = 604800; + +/// Process-local swap ID counter standing in for the contract's `NextId` +/// until the handlers are wired to a live Soroban RPC client. +static NEXT_SWAP_ID: AtomicU64 = AtomicU64::new(0); + +/// Current Unix timestamp in seconds (substitute for the ledger timestamp). +fn now_timestamp() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|duration| duration.as_secs()) + .unwrap_or(0) +} + // ── IP Registry ─────────────────────────────────────────────────────────────── /// Timestamp a new IP commitment. Returns the assigned IP ID. @@ -282,7 +301,7 @@ pub async fn initiate_swap(Json(body): Json) -> Result) -> } } - // TODO: Call Soroban RPC to invoke atomic_swap.batch_initiate_swap - // On success, cache the result: BATCH_SWAP_IDEMPOTENCY.insert(key, (json_value, Instant::now())); - Err(( - StatusCode::BAD_REQUEST, - Json(ErrorResponse { - error: "batch_initiate_swap not yet implemented".to_string(), - }), - )) + // #520: Cap the batch size at the contract's MAX_BATCH_SIZE (50). + if body.ip_ids.len() > MAX_BATCH_SIZE { + return Err(( + StatusCode::BAD_REQUEST, + Json(ErrorResponse { + error: format!( + "batch size {} exceeds maximum of {}", + body.ip_ids.len(), + MAX_BATCH_SIZE + ), + }), + )); + } + + // #520: Every price must be positive (contract: require_positive_price). + if let Some((&ip_id, _)) = body + .ip_ids + .iter() + .zip(body.prices.iter()) + .find(|(_, &price)| price <= 0) + { + return Err(( + StatusCode::BAD_REQUEST, + Json(ErrorResponse { + error: format!("price must be positive for ip_id {}", ip_id), + }), + )); + } + + // Reuse the batch semantics of the contract's `batch_initiate_swap` + // (validated in the contract's batch tests): every IP in the batch gets a + // fresh Pending swap with a ~7-day expiry, IDs allocated sequentially + // (contract NextId). Until the handlers are wired to a live Soroban RPC + // client, the records are served back through the #316 cache so + // GET /swap/{swap_id} can read them. + let expiry = now_timestamp() + SWAP_EXPIRY_SECONDS; + let mut swap_ids = Vec::with_capacity(body.ip_ids.len()); + for (&ip_id, &price) in body.ip_ids.iter().zip(body.prices.iter()) { + let swap_id = NEXT_SWAP_ID.fetch_add(1, Ordering::Relaxed); + let record = SwapRecord { + ip_id, + ip_registry_id: body.ip_registry_id.clone(), + seller: body.seller.clone(), + buyer: body.buyer.clone(), + price, + token: body.token.clone(), + status: SwapStatus::Pending, + expiry, + }; + cache::set_with_ttl(&cache::swap_key(swap_id), &record, SWAP_EXPIRY_SECONDS); + swap_ids.push(swap_id); + } + + let response = BatchInitiateSwapResponse { swap_ids }; + + // #523: Cache the result under the idempotency key so replays return the + // same swap IDs instead of allocating new ones. + if let Some(ref key) = body.idempotency_key { + BATCH_SWAP_IDEMPOTENCY.insert( + key.clone(), + (serde_json::to_value(&response).unwrap(), Instant::now()), + ); + } + + Ok(Json(response)) } /// Buyer accepts a pending swap. diff --git a/api-server/src/main.rs b/api-server/src/main.rs index f1b99e2..b47e88e 100644 --- a/api-server/src/main.rs +++ b/api-server/src/main.rs @@ -513,6 +513,147 @@ mod tests { assert_eq!(resp.status(), StatusCode::BAD_REQUEST); } + #[tokio::test] + async fn test_batch_initiate_swap_success_returns_pending_swaps() { + let app = build_app(); + let resp = app + .oneshot( + Request::builder() + .method("POST") + .uri("/v1/swap/bulk/initiate") + .header("content-type", "application/json") + .body(Body::from(r#"{"ip_registry_id":"C1","ip_ids":[1,2,3],"seller":"G1","prices":[100,200,300],"buyer":"G2","token":"C2"}"#)) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + let body = axum::body::to_bytes(resp.into_body(), usize::MAX).await.unwrap(); + let json: serde_json::Value = serde_json::from_slice(&body).unwrap(); + let swap_ids: Vec = json["swap_ids"] + .as_array() + .unwrap() + .iter() + .map(|id| id.as_u64().unwrap()) + .collect(); + assert_eq!(swap_ids.len(), 3); + // IDs are allocated sequentially, mirroring the contract's NextId. + assert_eq!(swap_ids[0] + 1, swap_ids[1]); + assert_eq!(swap_ids[1] + 1, swap_ids[2]); + + // The created swaps are readable via GET /swap/{id} as Pending with a + // ~7-day expiry, matching the contract's batch_initiate_swap. + for swap_id in swap_ids { + let resp = app + .clone() + .oneshot( + Request::builder() + .method("GET") + .uri(format!("/v1/swap/{}", swap_id)) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + let body = axum::body::to_bytes(resp.into_body(), usize::MAX).await.unwrap(); + let record: serde_json::Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(record["status"], "Pending"); + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs(); + assert!(record["expiry"].as_u64().unwrap() > now); + } + } + + #[tokio::test] + async fn test_batch_initiate_swap_too_large_returns_400() { + let app = build_app(); + let ip_ids: Vec = (1..=51).map(|i| i.to_string()).collect(); + let prices: Vec = (1..=51).map(|i| (i * 100).to_string()).collect(); + let body = format!( + r#"{{"ip_registry_id":"C1","ip_ids":[{}],"seller":"G1","prices":[{}],"buyer":"G2","token":"C2"}}"#, + ip_ids.join(","), + prices.join(",") + ); + let resp = app + .oneshot( + Request::builder() + .method("POST") + .uri("/v1/swap/bulk/initiate") + .header("content-type", "application/json") + .body(Body::from(body)) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::BAD_REQUEST); + let body = axum::body::to_bytes(resp.into_body(), usize::MAX).await.unwrap(); + let json: serde_json::Value = serde_json::from_slice(&body).unwrap(); + assert!(json["error"].as_str().unwrap().contains("exceeds maximum")); + } + + #[tokio::test] + async fn test_batch_initiate_swap_non_positive_price_returns_400() { + let app = build_app(); + let resp = app + .oneshot( + Request::builder() + .method("POST") + .uri("/v1/swap/bulk/initiate") + .header("content-type", "application/json") + .body(Body::from(r#"{"ip_registry_id":"C1","ip_ids":[1,2],"seller":"G1","prices":[100,0],"buyer":"G2","token":"C2"}"#)) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::BAD_REQUEST); + let body = axum::body::to_bytes(resp.into_body(), usize::MAX).await.unwrap(); + let json: serde_json::Value = serde_json::from_slice(&body).unwrap(); + assert!(json["error"].as_str().unwrap().contains("positive")); + } + + #[tokio::test] + async fn test_batch_initiate_swap_idempotent_replay_returns_same_ids() { + let app = build_app(); + let body = r#"{"ip_registry_id":"C1","ip_ids":[10,11],"seller":"G1","prices":[100,200],"buyer":"G2","token":"C2","idempotency_key":"batch-init-test-key"}"#; + let send = || { + app.clone().oneshot( + Request::builder() + .method("POST") + .uri("/v1/swap/bulk/initiate") + .header("content-type", "application/json") + .body(Body::from(body.to_string())) + .unwrap(), + ) + }; + + let first = send().await.unwrap(); + let second = send().await.unwrap(); + assert_eq!(first.status(), StatusCode::OK); + assert_eq!(second.status(), StatusCode::OK); + + let parse_ids = |resp: axum::response::Response| { + async move { + let body = axum::body::to_bytes(resp.into_body(), usize::MAX).await.unwrap(); + let json: serde_json::Value = serde_json::from_slice(&body).unwrap(); + json["swap_ids"] + .as_array() + .unwrap() + .iter() + .map(|id| id.as_u64().unwrap()) + .collect::>() + } + }; + + let first_ids = parse_ids(first).await; + let second_ids = parse_ids(second).await; + assert_eq!(first_ids.len(), 2); + // #523: replay with the same key must return the cached swap IDs. + assert_eq!(first_ids, second_ids); + } + // ── #319: API Versioning tests ──────────────────────────────────────────── #[tokio::test]