AmbireTech · elpiel · Mar 24, 2022 · Mar 1, 2022 · Mar 2, 2022 · Mar 4, 2022
diff --git a/adview-manager/serve/src/main.rs b/adview-manager/serve/src/main.rs
@@ -13,18 +13,19 @@ use wiremock::{
     Mock, MockServer, ResponseTemplate,
 };
 
-
-use tera::{Tera, Context};
+use tera::{Context, Tera};
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     env_logger::init();
 
     let serve_dir = match std::env::current_dir().unwrap() {
         serve_path if serve_path.ends_with("serve") => serve_path,
-        adview_manager_path if adview_manager_path.ends_with("adview-manager") => adview_manager_path.join("serve"),
+        adview_manager_path if adview_manager_path.ends_with("adview-manager") => {
+            adview_manager_path.join("serve")
+        }
         // running from the Validator stack workspace
-        workspace_path => workspace_path.join("adview-manager/serve")
+        workspace_path => workspace_path.join("adview-manager/serve"),
     };
 
     let templates_glob = format!("{}/templates/**/*.html", serve_dir.display());

diff --git a/primitives/src/sentry.rs b/primitives/src/sentry.rs
@@ -3,7 +3,7 @@ use crate::{
     balances::BalancesState,
     spender::Spender,
     validator::{ApproveState, Heartbeat, MessageTypes, NewState, Type as MessageType},
-    Address, Balances, CampaignId, UnifiedNum, ValidatorId, IPFS,
+    Address, Balances, CampaignId, UnifiedMap, UnifiedNum, ValidatorId, IPFS,
 };
 use chrono::{
     serde::ts_milliseconds, Date, DateTime, Datelike, Duration, NaiveDate, TimeZone, Timelike, Utc,
@@ -590,6 +590,11 @@ pub struct AllSpendersQuery {
     pub page: u64,
 }
 
+#[derive(Debug, Serialize, Deserialize)]
+pub struct ChannelPayRequest {
+    pub payouts: UnifiedMap,
+}
+
 #[derive(Serialize, Deserialize, Debug)]
 pub struct ValidatorMessage {
     pub from: ValidatorId,

diff --git a/sentry/src/db/channel.rs b/sentry/src/db/channel.rs
@@ -1,4 +1,4 @@
-use primitives::{Channel, ChannelId, ValidatorId};
+use primitives::{Channel, ChannelId};
 
 pub use list_channels::list_channels;
 

diff --git a/sentry/src/lib.rs b/sentry/src/lib.rs
@@ -26,8 +26,8 @@ use {
         campaign,
         campaign::{campaign_list, create_campaign, update_campaign},
         channel::{
-            add_spender_leaf, channel_list, get_accounting_for_channel, get_all_spender_limits,
-            get_spender_limits, last_approved,
+            add_spender_leaf, channel_list, channel_payout, get_accounting_for_channel,
+            get_all_spender_limits, get_spender_limits, last_approved,
             validator_message::{
                 create_validator_messages, extract_params, list_validator_messages,
             },
@@ -77,6 +77,9 @@ static CHANNEL_ACCOUNTING: Lazy<Regex> = Lazy::new(|| {
     Regex::new(r"^/v5/channel/0x([a-zA-Z0-9]{64})/accounting/?$")
         .expect("The regex should be valid")
 });
+static CHANNEL_PAY: Lazy<Regex> = Lazy::new(|| {
+    Regex::new(r"^/v5/channel/0x([a-zA-Z0-9]{64})/pay/?$").expect("The regex should be valid")
+});
 
 /// Regex extracted parameters.
 /// This struct is created manually on each of the matched routes.
@@ -210,10 +213,6 @@ async fn campaigns_router<C: Locked + 'static>(
 }
 
 // TODO AIP#61: Add routes for:
-// - POST /channel/:id/pay
-// #[serde(rename_all = "camelCase")]
-// Pay { payout: BalancesMap },
-//
 // - GET /channel/:id/get-leaf
 async fn channels_router<C: Locked + 'static>(
     mut req: Request<Body>,
@@ -345,6 +344,21 @@ async fn channels_router<C: Locked + 'static>(
             .await?;
 
         get_accounting_for_channel(req, app).await
+    }
+    // POST /v5/channel/:id/pay
+    else if let (Some(caps), &Method::POST) = (CHANNEL_PAY.captures(&path), method) {
+        let param = RouteParams(vec![caps
+            .get(1)
+            .map_or("".to_string(), |m| m.as_str().to_string())]);
+        req.extensions_mut().insert(param);
+
+        req = Chain::new()
+            .chain(AuthRequired)
+            .chain(ChannelLoad)
+            .apply(req, app)
+            .await?;
+
+        channel_payout(req, app).await
     } else {
         Err(ResponseError::NotFound)
     }

diff --git a/sentry/src/routes.rs b/sentry/src/routes.rs
@@ -76,13 +76,19 @@
 //!
 //! Response: [`LastApprovedResponse`][primitives::sentry::LastApprovedResponse]
 //!
-//! - `POST /v5/channel/:id/pay` (auth required)
+//! - [`POST /v5/channel/:id/pay`](channel::channel_payout) (auth required)
 //!
-//! TODO: implement and document as part of issue #382
+//! Channel Payout with authentication of the spender.
+//!
+//! This route handles withdrawals of advertiser funds for the authenticated spender.
+//! It needs to ensure all campaigns are closed. It accepts a JSON body in the request which contains
+//! all of the earners and updates their balances accordingly. Used when an advertiser/spender wants
+//! to get their remaining funds back.
+//!
+//! Request JSON body: [`ChannelPayRequest`](primitives::sentry::ChannelPayRequest)
 //!
-//! Channel Payout with authentication of the spender
+//! Response: [`SuccessResponse`](primitives::sentry::SuccessResponse)
 //!
-//! Withdrawals of advertiser funds - re-introduces the PAY event with a separate route.
 //!
 //! - `GET /v5/channel/:id/get-leaf`
 //!

diff --git a/sentry/src/routes/campaign.rs b/sentry/src/routes/campaign.rs
@@ -14,7 +14,7 @@ use crate::{
 };
 use adapter::{prelude::*, Adapter, Error as AdaptorError};
 use deadpool_postgres::PoolError;
-use futures::future::try_join_all;
+use futures::{future::try_join_all, TryFutureExt};
 use hyper::{Body, Request, Response};
 use primitives::{
     campaign_validator::Validator,
@@ -115,17 +115,24 @@ pub async fn fetch_campaign_ids_for_channel(
     if total_pages < 2 {
         Ok(campaign_ids)
     } else {
-        let other_pages: Vec<Vec<CampaignId>> = try_join_all((1..total_pages).map(|i| {
-            get_campaign_ids_by_channel(
-                pool,
-                &channel_id,
-                limit.into(),
-                i.checked_mul(limit.into()).expect("TODO"),
-            )
+        let pages_skip: Vec<u64> = (1..total_pages)
+            .map(|i| {
+                i.checked_mul(limit.into()).ok_or_else(|| {
+                    ResponseError::FailedValidation(
+                        "Calculating skip while fetching campaign ids results in an overflow"
+                            .to_string(),
+                    )
+                })
+            })
+            .collect::<Result<_, _>>()?;
+
+        let other_pages = try_join_all(pages_skip.into_iter().map(|skip| {
+            get_campaign_ids_by_channel(pool, &channel_id, limit.into(), skip)
+                .map_err(|e| ResponseError::BadRequest(e.to_string()))
         }))
         .await?;
 
-        let all_campaigns: Vec<CampaignId> = std::iter::once(campaign_ids)
+        let all_campaigns = std::iter::once(campaign_ids)
             .chain(other_pages.into_iter())
             .flat_map(|campaign_ids| campaign_ids.into_iter())
             .collect();
@@ -1279,13 +1286,19 @@ mod test {
             let new_budget = UnifiedNum::from_u64(300 * multiplier);
             let delta_budget = get_delta_budget(&campaign_remaining, &campaign, new_budget).await;
 
-            assert!(matches!(&delta_budget, Err(Error::NewBudget(_))), "Got result: {delta_budget:?}");
+            assert!(
+                matches!(&delta_budget, Err(Error::NewBudget(_))),
+                "Got result: {delta_budget:?}"
+            );
 
             // campaign_spent == new_budget
             let new_budget = UnifiedNum::from_u64(400 * multiplier);
             let delta_budget = get_delta_budget(&campaign_remaining, &campaign, new_budget).await;
 
-            assert!(matches!(&delta_budget, Err(Error::NewBudget(_))), "Got result: {delta_budget:?}");
+            assert!(
+                matches!(&delta_budget, Err(Error::NewBudget(_))),
+                "Got result: {delta_budget:?}"
+            );
         }
         // Increasing budget
         {