Merge pull request #947 from okonkwofreeman001/feat/wire-batch-initiate-endpoint

feat(api-server): wire batch-initiate endpoint to contract batch semantics

Author: fejilaup-cloud

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;
@@ -19,6 +20,24 @@ use crate::webhook;
 // #523: Per-handler idempotency store for batch swap operations.
 static BATCH_SWAP_IDEMPOTENCY: Lazy<DeduplicationStore> = 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.
@@ -307,7 +326,7 @@ pub async fn initiate_swap(Json(body): Json<InitiateSwapRequest>) -> Result<Json
     request_body = BatchInitiateSwapRequest,
     responses(
         (status = 200, description = "Swaps initiated, returns swap_ids", body = BatchInitiateSwapResponse),
-        (status = 400, description = "Validation error (mismatched lengths, invalid IP, etc.)", body = ErrorResponse),
+        (status = 400, description = "Validation error (mismatched lengths, empty/oversized batch, non-positive price, duplicate ip_ids, etc.)", body = ErrorResponse),
     )
 )]
 #[instrument(skip(body))]
@@ -358,14 +377,71 @@ pub async fn batch_initiate_swap(Json(body): Json<BatchInitiateSwapRequest>) ->
         }
     }
 
-    // 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.

api-server/src/main.rs

@@ -530,6 +530,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<u64> = 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<String> = (1..=51).map(|i| i.to_string()).collect();
+        let prices: Vec<String> = (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::<Vec<u64>>()
+            }
+        };
+
+        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]