Enable amending allocations

Cargo.toml

@@ -23,7 +23,7 @@ packet-trace-enable = [
     "ya-vpn/packet-trace-enable",
     "ya-file-logging/packet-trace-enable",
     "ya-net/packet-trace-enable",
-    "ya-service-bus/packet-trace-enable"
+    "ya-service-bus/packet-trace-enable",
 ]
 
 [[bin]]
@@ -216,7 +216,7 @@ members = [
     "utils/fd-metrics",
     "core/metrics",
     "test-utils/test-framework",
-    "test-utils/test-framework/framework-macro"
+    "test-utils/test-framework/framework-macro",
 ]
 
 [patch.crates-io]
@@ -245,19 +245,19 @@ ya-service-api-interfaces = { path = "core/serv-api/interfaces" }
 ya-service-api-web = { path = "core/serv-api/web" }
 
 ## SERVICE BUS
-ya-service-bus = { git = "https://github.com/golemfactory/ya-service-bus.git", rev = "190f0d772f7ed0830d54a2cef77d7a177f276c68"}
-ya-sb-proto = { git = "https://github.com/golemfactory/ya-service-bus.git", rev = "190f0d772f7ed0830d54a2cef77d7a177f276c68"}
-ya-sb-router = { git = "https://github.com/golemfactory/ya-service-bus.git", rev = "190f0d772f7ed0830d54a2cef77d7a177f276c68"}
-ya-sb-util = { git = "https://github.com/golemfactory/ya-service-bus.git", rev = "190f0d772f7ed0830d54a2cef77d7a177f276c68"}
+ya-service-bus = { git = "https://github.com/golemfactory/ya-service-bus.git", rev = "190f0d772f7ed0830d54a2cef77d7a177f276c68" }
+ya-sb-proto = { git = "https://github.com/golemfactory/ya-service-bus.git", rev = "190f0d772f7ed0830d54a2cef77d7a177f276c68" }
+ya-sb-router = { git = "https://github.com/golemfactory/ya-service-bus.git", rev = "190f0d772f7ed0830d54a2cef77d7a177f276c68" }
+ya-sb-util = { git = "https://github.com/golemfactory/ya-service-bus.git", rev = "190f0d772f7ed0830d54a2cef77d7a177f276c68" }
 
 #ya-service-bus = { path = "../ya-service-bus" }
 #ya-sb-proto = { path = "../ya-service-bus/crates/proto" }
 #ya-sb-router = { path = "../ya-service-bus/crates/router" }
 #ya-sb-util = { path = "../ya-service-bus/crates/util" }
 
 ## CLIENT
-#ya-client = { git = "https://github.com/golemfactory/ya-client.git", rev = "2a6350f62cf8d926721225a3451822731456e3fe" }
-#ya-client-model = { git = "https://github.com/golemfactory/ya-client.git", rev = "2a6350f62cf8d926721225a3451822731456e3fe" }
+ya-client = { git = "https://github.com/golemfactory/ya-client.git", rev = "76f4b6c98ca8c3ce73bae2dc66531906dd985b2d" }
+ya-client-model = { git = "https://github.com/golemfactory/ya-client.git", rev = "76f4b6c98ca8c3ce73bae2dc66531906dd985b2d" }
 
 ## RELAY and networking stack
 ya-relay-stack = { git = "https://github.com/golemfactory/ya-relay.git", rev = "c92a75b0cf062fcc9dbb3ea2a034d913e5fad8e5" }
@@ -292,7 +292,7 @@ ethereum-tx-sign = { git = "https://github.com/mfranciszkiewicz/ethereum-tx-sign
 graphene-sgx = { git = " https://github.com/golemfactory/graphene-rust.git", rev = "dbd993ebad7f9190410ea390a589348479af6407" }
 web3 = { git = "https://github.com/golemfactory/rust-web3", branch = "update_ethabi" }
 
-diesel = { git = "https://github.com/golemfactory/yagna-diesel-patch.git", rev = "a512c66d520a9066dd9a4d1416f9109019b39563"}
+diesel = { git = "https://github.com/golemfactory/yagna-diesel-patch.git", rev = "a512c66d520a9066dd9a4d1416f9109019b39563" }
 
 # Speed up builds on macOS (will be default in next rust version probably)
 # https://jakedeichert.com/blog/reducing-rust-incremental-compilation-times-on-macos-by-70-percent/

core/payment/src/api/allocations.rs

@@ -3,12 +3,14 @@
 // External crates
 use actix_web::web::{delete, get, post, put, Data, Json, Path, Query};
 use actix_web::{HttpResponse, Scope};
+use bigdecimal::BigDecimal;
 use chrono::{DateTime, Utc};
 use serde_json::value::Value::Null;
 use ya_client_model::NodeId;
 
 // Workspace uses
 use ya_agreement_utils::{ClauseOperator, ConstraintKey, Constraints};
+use ya_client_model::payment::allocation::AllocationUpdate;
 use ya_client_model::payment::*;
 use ya_core_model::payment::local::{
     ValidateAllocation, ValidateAllocationError, BUS_ID as LOCAL_SERVICE,
@@ -162,12 +164,94 @@
     }
 }
 
+fn amend_allocation_fields(
+    old_allocation: Allocation,
+    update: AllocationUpdate,
+) -> Result<Allocation, &'static str> {
+    let total_amount = update
+        .total_amount
+        .unwrap_or_else(|| old_allocation.total_amount.clone());
+    let remaining_amount = total_amount.clone() - &old_allocation.spent_amount;
+
+    if remaining_amount < 0 {
+        return Err("New allocation would be smaller than the already spent amount");
+    }
+    if let Some(timeout) = update.timeout {
+        if timeout < chrono::offset::Utc::now() {
+            return Err("New allocation timeout is in the past");
+        }
+    }
+
+    Ok(Allocation {
+        total_amount,
+        remaining_amount,
+        timeout: update.timeout.or(old_allocation.timeout),
+        ..old_allocation
+    })
+}
+
 async fn amend_allocation(
     db: Data<DbExecutor>,
     path: Path<params::AllocationId>,
-    body: Json<Allocation>,
+    body: Json<AllocationUpdate>,
+    id: Identity,
 ) -> HttpResponse {
-    response::not_implemented() // TODO
+    let allocation_id = path.allocation_id.clone();
+    let node_id = id.identity;
+    let new_allocation: AllocationUpdate = body.into_inner();
+    let dao: AllocationDao = db.as_dao();
+
+    let current_allocation = match dao.get(allocation_id.clone(), node_id).await {
+        Ok(AllocationStatus::Active(allocation)) => allocation,
+        Ok(AllocationStatus::Gone) => {
+            return response::gone(&format!(
+                "Allocation {allocation_id} has been already released",
+            ))
+        }
+        Ok(AllocationStatus::NotFound) => return response::not_found(),
+        Err(e) => return response::server_error(&e),
+    };
+
+    let amended_allocation =
+        match amend_allocation_fields(current_allocation.clone(), new_allocation) {
+            Ok(allocation) => allocation,
+            Err(e) => return response::bad_request(&e),
+        };
+
+    // validation will take into account all existing allocation, including the one
+    // being currently modified. This means we only need to validate the increase.
+    let amount_to_validate =
+        amended_allocation.total_amount.clone() - &current_allocation.total_amount;
+
+    let validate_msg = ValidateAllocation {
+        platform: amended_allocation.payment_platform.clone(),
+        address: amended_allocation.address.clone(),
+        amount: if &amount_to_validate > &BigDecimal::from(0) {
+            amount_to_validate
+        } else {
+            0.into()
+        },
+    };
+    match async move { Ok(bus::service(LOCAL_SERVICE).send(validate_msg).await??) }.await {
+        Ok(true) => {}
+        Ok(false) => return response::bad_request(&"Insufficient funds to make allocation. Top up your account or release all existing allocations to unlock the funds via `yagna payment release-allocations`"),
+        Err(Error::Rpc(RpcMessageError::ValidateAllocation(
+                           ValidateAllocationError::AccountNotRegistered,
+                       ))) => return response::bad_request(&"Account not registered"),
+        Err(e) => return response::server_error(&e),
+    }
+
+    match dao.replace(amended_allocation, node_id).await {
+        Ok(true) => {}
+        Ok(false) => {
+            return response::server_error(
+                &"Allocation not present despite preconditions being already ensured",
+            )
+        }
+        Err(e) => return response::server_error(&e),
+    }
+
+    get_allocation(db, path, id).await
 }
 
 async fn release_allocation(

core/payment/src/dao/allocation.rs

@@ -63,6 +63,20 @@ impl<'c> AllocationDao<'c> {
         .await
     }
 
+    pub async fn replace(&self, allocation: Allocation, owner_id: NodeId) -> DbResult<bool> {
+        do_with_transaction(self.pool, move |conn| {
+            let count = diesel::update(dsl::pay_allocation)
+                .filter(dsl::id.eq(allocation.allocation_id.clone()))
+                .filter(dsl::owner_id.eq(&owner_id))
+                .filter(dsl::released.eq(false))
+                .set(WriteObj::from_allocation(allocation, owner_id))
+                .execute(conn)?;
+
+            Ok(count == 1)
+        })
+        .await
+    }
+
     pub async fn get(&self, allocation_id: String, owner_id: NodeId) -> DbResult<AllocationStatus> {
         readonly_transaction(self.pool, move |conn| {
             let allocation: Option<ReadObj> = dsl::pay_allocation

core/payment/src/models/allocation.rs

@@ -5,7 +5,7 @@ use ya_client_model::payment::{Allocation, NewAllocation};
 use ya_client_model::NodeId;
 use ya_persistence::types::BigDecimalField;
 
-#[derive(Queryable, Debug, Identifiable, Insertable)]
+#[derive(Queryable, Debug, Identifiable, Insertable, AsChangeset)]
 #[table_name = "pay_allocation"]
 pub struct WriteObj {
     pub id: String,
@@ -56,6 +56,21 @@ impl WriteObj {
             released: false,
         }
     }
+
+    pub fn from_allocation(allocation: Allocation, owner_id: NodeId) -> Self {
+        Self {
+            id: allocation.allocation_id,
+            owner_id,
+            payment_platform: allocation.payment_platform,
+            address: allocation.address,
+            total_amount: allocation.total_amount.into(),
+            spent_amount: allocation.spent_amount.into(),
+            remaining_amount: allocation.remaining_amount.into(),
+            timeout: allocation.timeout.map(|v| v.naive_utc()),
+            make_deposit: allocation.make_deposit,
+            released: false,
+        }
+    }
 }
 
 impl From<ReadObj> for Allocation {