Merge pull request #58 from Cerebellum-Network/feature/bucket-ownership

Feature/update-payments-and-commits

bucket/ddc_bucket/buckets_perms/store.rs

@@ -27,7 +27,7 @@ impl BucketsPermsStore {
 
     // get accounts with permission for bucket writing
     pub fn get_bucket_writers(&self, key: BucketId) -> Vec<AccountId> {
-        let writers = (*self.writers.get(&key).unwrap()).clone();
+        let writers = (*self.writers.get(&key).unwrap_or(&Vec::new())).clone();
         return writers;
     }
 
@@ -54,7 +54,7 @@ impl BucketsPermsStore {
 
     // get accounts with permission for bucket reading
     pub fn get_bucket_readers(&self, key: BucketId) -> Vec<AccountId> {
-        let readers = (*self.readers.get(&key).unwrap()).clone();
+        let readers = (*self.readers.get(&key).unwrap_or(&Vec::new())).clone();
         return readers;
     }
 

bucket/ddc_bucket/cdn_cluster/entity.rs

@@ -4,7 +4,7 @@ use ink_prelude::vec::Vec;
 use ink_storage::traits::{PackedLayout, SpreadLayout};
 use scale::{Decode, Encode};
 
-use crate::ddc_bucket::{AccountId, NodeId, Result};
+use crate::ddc_bucket::{AccountId, Balance, NodeId, Result};
 use crate::ddc_bucket::cash::{Cash, Payable};
 use crate::ddc_bucket::contract_fee::{SIZE_ACCOUNT_ID, SIZE_BALANCE, SIZE_PER_RECORD, SIZE_RESOURCE, SIZE_VEC};
 use crate::ddc_bucket::Error::{UnauthorizedClusterManager, InsufficientBalance};
@@ -23,7 +23,7 @@ pub struct CdnCluster {
     pub cdn_nodes: Vec<NodeId>,
     pub resources_used: Resource,
     pub revenues: Cash,
-    pub usd_per_gb: u128,
+    pub usd_per_gb: Balance,
 }
 
 #[derive(Clone, PartialEq, Encode, Decode)]
@@ -41,7 +41,8 @@ impl CdnCluster {
         CdnCluster {
             manager_id,
             cdn_nodes,
-            usd_per_gb: 0,
+            // usd_per_gb: 100_000_000, // setting initially to 1 cent per GB
+            usd_per_gb: 104_857_600, // setting initially to 1 cent per GB
             resources_used: 0,
             revenues: Cash(0),
         }
@@ -60,11 +61,15 @@ impl CdnCluster {
         self.revenues
     }
 
-    pub fn set_rate(&mut self, usd_per_gb: u128) -> Result<()> {
+    pub fn set_rate(&mut self, usd_per_gb: Balance) -> Result<()> {
         self.usd_per_gb = usd_per_gb;
         Ok(())
     }
 
+    pub fn get_rate(&self) -> Balance {
+        self.usd_per_gb
+    }
+
     pub fn put_revenues(&mut self, amount: Cash) {
         self.revenues.increase(amount);
     }
@@ -81,5 +86,3 @@ impl CdnCluster {
         if self.manager_id == caller { Ok(()) } else { Err(UnauthorizedClusterManager) }
     }
 }
-
-pub const USD_PER_GB: u128 = 1;

bucket/ddc_bucket/cdn_cluster/messages.rs

@@ -11,6 +11,8 @@ use crate::ddc_bucket::cdn_node::entity::{CdnNode, NodeId, Resource};
 use crate::ddc_bucket::perm::entity::Permission;
 use crate::ddc_bucket::perm::store::PermStore;
 
+const KB_PER_GB: Balance = 1_000_000;
+
 impl DdcBucket {
     pub fn message_cdn_cluster_create(
         &mut self,
@@ -46,15 +48,32 @@ impl DdcBucket {
         Ok(())
     }
 
+    // Get the price usd per gb
+    pub fn message_cdn_get_rate(&self, cluster_id: ClusterId) -> Result<Balance> {
+        let cluster = self.cdn_clusters.get(cluster_id)?;
+        let rate = cluster.get_rate();
+        Ok(rate)
+    }
+
     // First payment is for aggregate consumption for account, second is the aggregate payment for the node (u32 for ids)
     pub fn message_cdn_cluster_put_revenue(&mut self, cluster_id: ClusterId, aggregates_accounts: Vec<(AccountId, u128)>, aggregates_nodes: Vec<(u32, u128)>, aggregates_buckets: Vec<(BucketId, Resource)>) -> Result<()> {
         let cluster = self.cdn_clusters.get_mut(cluster_id)?;
-        Self::only_cdn_cluster_manager(cluster)?;
+        // Self::only_cdn_cluster_manager(cluster)?;
 
         let mut cluster_payment = 0;
         let mut _undistributed_payment_accounts = 0;
 
-        for &(client_id, payment) in aggregates_accounts.iter() {
+        let aggregate_payments_accounts;
+        {
+            let conv = &self.accounts.1;
+            aggregate_payments_accounts = aggregates_accounts.iter().map(|(client_id, resources_used)| {
+                let account_id = *client_id;
+                let cere_payment: Balance = conv.to_cere(*resources_used as Balance * cluster.usd_per_gb / KB_PER_GB );
+                (account_id, cere_payment)
+            }).collect::<Vec<(AccountId, Balance)>>();
+        }
+
+        for &(client_id, payment) in aggregate_payments_accounts.iter() {
             let account = self.accounts.0.get_mut(&client_id)
             .ok_or(InsufficientBalance)?;
             account.withdraw_bonded(Payable(payment))?;
@@ -66,7 +85,7 @@ impl DdcBucket {
         for &(node_id, resources_used) in aggregates_nodes.iter() {
             let node = self.cdn_nodes.get_mut(node_id)?;
 
-            let payment = conv.to_cere (resources_used as Balance * cluster.usd_per_gb / 1000000 );
+            let payment = conv.to_cere (resources_used as Balance * cluster.usd_per_gb / KB_PER_GB );
 
             node.put_payment(payment);
             cluster_payment += payment;

bucket/ddc_bucket/committer/messages.rs

@@ -1,30 +1,42 @@
 use ink_lang::{StaticEnv};
 use ink_env::Error;
+use ink_prelude::vec::Vec;
 
-use crate::ddc_bucket::{AccountId, DdcBucket};
-use super::store::{Commit, EraConfig};
+use crate::ddc_bucket::{AccountId, DdcBucket, NodeId};
+use super::store::{Commit, EraConfig, EraStatus, EraAndTimestamp};
 
 pub type Result<T> = core::result::Result<T, Error>;
 
 impl DdcBucket {
-    pub fn message_get_commit(&self, node: AccountId) -> Commit {
-        self.committer_store.get_commit(node)
+    pub fn message_set_commit(&mut self, cdn_owner: AccountId, node_id: NodeId, commit: Commit) {
+        self.committer_store.set_commit(cdn_owner, node_id, commit);
     }
 
-    pub fn message_set_commit(&mut self, node: AccountId, commit: Commit) {
-        self.committer_store.set_commit(node, commit);
+    pub fn message_get_commit(&self, cdn_owner: AccountId) -> Vec<(NodeId, Commit)> {
+        self.committer_store.get_commit(cdn_owner)
+    }
+
+    pub fn message_set_validated_commit(&mut self, node: AccountId, era: u64) -> Result<()> {
+        match self.committer_store.set_validated_commit(node, era) {
+            Err(_e) => panic!("Setting validated commit failed"), 
+            Ok(_v) => Ok(()),
+        }
+    }
+
+    pub fn message_get_validated_commit(&self, node: AccountId) -> EraAndTimestamp {
+        self.committer_store.get_validate_commit(node)
     }
 
     pub fn message_set_era(&mut self, era_config: EraConfig) -> Result<()> {
         let caller = Self::env().caller();
 
         match self.committer_store.set_era(caller, era_config) {
-          Err(_e) => panic!("Setting erra failed"), 
-          Ok(_v) => Ok(()),
+            Err(_e) => panic!("Setting era failed"), 
+            Ok(_v) => Ok(()),
         }
-    }
+    }  
 
-    pub fn message_get_era(&self) -> u64 {
+    pub fn message_get_era(&self) -> EraStatus {
         let timestamp = Self::env().block_timestamp();
         self.committer_store.get_era(timestamp)
     }

bucket/ddc_bucket/committer/store.rs

@@ -1,11 +1,12 @@
-use crate::ddc_bucket::{AccountId, Hash};
+use crate::ddc_bucket::{AccountId, Hash, NodeId};
 
 use ink_storage::{
     collections::{
         HashMap as StorageHashMap,
     },
     traits,
 };
+use ink_prelude::vec::Vec;
 
 #[derive(Debug, PartialEq, scale::Encode, scale::Decode)]
 #[cfg_attr(feature = "std", derive(scale_info::TypeInfo))]
@@ -14,9 +15,26 @@ pub enum Error {
     UnauthorizedOperator
 }
 
-pub type Result<T> = core::result::Result<T, Error>;
+/// Within the concept of era we would like to return specific phase to interested agents
+#[derive(Debug, PartialEq, scale::Encode, scale::Decode)]
+#[cfg_attr(feature = "std", derive(scale_info::TypeInfo))]
+pub enum Phase {
+    Commit,
+    Valiadation,
+    Payout
+}
 
-#[derive(Copy, Clone, traits::PackedLayout, traits::SpreadLayout, scale::Encode, scale::Decode)]
+#[derive(Debug, PartialEq, scale::Encode, scale::Decode)]
+#[cfg_attr(feature = "std", derive(scale_info::TypeInfo))]
+pub struct  EraStatus {
+    current_era: u64,
+    current_phase: Phase,
+    previous_era: u64,
+    prev_era_from_timestamp: u64,
+    prev_era_to_timestamp: u64,
+}
+
+#[derive(Copy, Clone, traits::PackedLayout, traits::SpreadLayout, scale::Encode, scale::Decode, Debug)]
 #[cfg_attr(feature = "std", derive(::scale_info::TypeInfo))]
 pub struct Commit {
     hash: Hash,
@@ -25,45 +43,73 @@ pub struct Commit {
     to_timestamp: u64,
 } 
 
-#[derive(Copy, Clone, traits::SpreadLayout, scale::Encode, scale::Decode)]
+#[derive(Copy, Clone, traits::SpreadLayout, scale::Encode, scale::Decode, Debug)]
 #[cfg_attr(feature = "std", derive(::scale_info::TypeInfo, traits::StorageLayout))]
 pub struct EraConfig {
     start: u64,
     interval: u64,
-    commit_deadline: u64,
-    validation_deadline:u64
+    commit_duration: u64,
+    validation_duration:u64
 }  
 
-#[derive(traits::SpreadLayout)]
+#[derive(traits::SpreadLayout, Debug)]
 #[cfg_attr(feature = "std", derive(traits::StorageLayout))]
 pub struct CommitterStore {
     operator_id: AccountId,
-    commits: StorageHashMap<AccountId, Commit>,
+    commits: StorageHashMap<AccountId, Vec<(NodeId, Commit)>>,
+    validated_commits: StorageHashMap<AccountId, EraAndTimestamp>,
     era_settings: EraConfig
 }
 
+pub type Result<T> = core::result::Result<T, Error>;
+
+pub type EraAndTimestamp = (u64, u64);
+
 impl CommitterStore {
     pub fn new(operator_id: AccountId) -> CommitterStore {
         CommitterStore {
             operator_id,
             commits: Default::default(),
+            validated_commits: Default::default(),
             era_settings: EraConfig {
                 start: 0,
                 interval: 0,
-                commit_deadline: 0,
-                validation_deadline: 0,
+                commit_duration: 0,
+                validation_duration: 0,
             }
         }
     }
 
-    pub fn get_commit(&self, node: AccountId) -> Commit {
-        *self.commits.get(&node).unwrap()
-    }
-
     /// The node can set the latest commit with this function
     /// check the sender !!!!
-    pub fn set_commit(&mut self, node: AccountId, commit: Commit) {
-        self.commits.insert(node, commit);
+    pub fn set_commit(&mut self, cdn_owner: AccountId, node_id: NodeId, commit: Commit) {
+        if !self.commits.contains_key(&cdn_owner) {
+            let empty_vec = Vec::<(u32, Commit)>::new();
+            self.commits.insert(cdn_owner, empty_vec);
+        }
+
+        let account_commits = self.commits.get_mut(&cdn_owner).unwrap();
+        let index = account_commits.iter().position(|(node, _)| *node == node_id).unwrap_or(usize::MAX);
+        if index != usize::MAX {
+            account_commits.remove(index);
+        }
+        account_commits.push((node_id, commit));
+    }
+
+    pub fn get_commit(&self, cdn_owner: AccountId) -> Vec<(NodeId, Commit)> {
+        self.commits.get(&cdn_owner).unwrap_or(&Vec::<(u32, Commit)>::new()).clone()
+    }
+
+    // Set the last validated commit per CDN node
+    pub fn set_validated_commit(&mut self, node: AccountId, era: u64) -> Result<()> {
+        let prev_era_to_timestamp = self.era_settings.start + self.era_settings.interval * (era + 1);
+        self.validated_commits.insert(node, (era, prev_era_to_timestamp));
+        Ok(())
+    }
+
+    // Get the last era & timestamp validated per CDN node
+    pub fn get_validate_commit(&self, node: AccountId) -> EraAndTimestamp {
+        *self.validated_commits.get(&node).unwrap_or(&(0,0))
     }
 
     // Akin to modifier
@@ -79,21 +125,33 @@ impl CommitterStore {
     }
 
     // Get the current era phase. 0 represents commit phase, 1 validation phase, 2 payout phase
-    pub fn get_era(&self, timestamp: u64) -> u64 {
+    pub fn get_era(&self, timestamp: u64) -> EraStatus {
         let era_start = self.era_settings.start;
         let interval = self.era_settings.interval;
         let elapsed_time_within_interval = (timestamp - era_start) % interval;
 
-        if elapsed_time_within_interval < self.era_settings.commit_deadline {
-            return 0;
-        } else if elapsed_time_within_interval < self.era_settings.validation_deadline {
-            return 1;
+        let current_phase = if elapsed_time_within_interval < self.era_settings.commit_duration {
+            Phase::Commit
+        } else if elapsed_time_within_interval < self.era_settings.validation_duration {
+            Phase::Valiadation
         } else {
-            return 2;
+            Phase::Payout
+        };
+
+        let current_era = (timestamp - era_start) / interval;
+        let previous_era = current_era - 1;
+        let prev_era_from_timestamp = era_start + interval * previous_era;
+        let prev_era_to_timestamp = era_start + interval * current_era;
+
+        EraStatus { 
+            current_era, 
+            current_phase, 
+            previous_era, 
+            prev_era_from_timestamp, 
+            prev_era_to_timestamp
         }
     }
 
-
     // Get the current era phase
     pub fn get_era_settings(&self) -> EraConfig {
         return self.era_settings;

bucket/ddc_bucket/currency.rs

@@ -12,15 +12,15 @@ pub type USD = Balance;
 #[cfg_attr(feature = "std", derive(traits::StorageLayout, Debug))]
 pub struct CurrencyConverter(Balance /* how many USD for PRECISION CERE */);
 
-const PRECISION: Balance = 10_000;
+const PRECISION: Balance = 10_000_000; 
 
 impl Default for CurrencyConverter {
     fn default() -> Self {
         Self(PRECISION)
     }
 }
 
-impl CurrencyConverter {
+impl CurrencyConverter {                        // 10_000_000
     pub fn set_usd_per_cere(&mut self, usd_per_cere: USD) {
         self.0 = usd_per_cere * PRECISION / TOKEN;
     }
@@ -32,4 +32,4 @@ impl CurrencyConverter {
     pub fn to_usd(&self, cere: CERE) -> USD {
         self.0 * cere / PRECISION
     }
-}
+}

bucket/ddc_bucket/tests/as_user.rs

@@ -1,4 +1,5 @@
 use crate::ddc_bucket::*;
+use crate::ddc_bucket::Error::BucketDoesNotExist;
 
 use super::as_gateway::GATEWAY_ENGINE;
 use super::as_storage::STORAGE_ENGINE;
@@ -30,7 +31,7 @@ impl TestUser {
         let cluster_id = find_cluster(contract, engine_name)?.cluster_id;
 
         push_caller_value(account_id, CONTRACT_FEE_LIMIT);
-        let bucket_id = contract.bucket_create(BUCKET_PARAMS.to_string().unwrap(), cluster_id);
+        let bucket_id = contract.bucket_create(BUCKET_PARAMS.to_string().unwrap(), cluster_id, None);
         pop_caller();
 
         // Allocate the bucket to the cluster.

bucket/ddc_bucket/tests/env_utils.rs

@@ -90,6 +90,9 @@ fn _print_events(events: &[Event]) {
             Event::Deposit(ev) => println!("EVENT {:?}", ev),
             Event::GrantPermission(ev) => println!("EVENT {:?}", ev),
             Event::RevokePermission(ev) => println!("EVENT {:?}", ev),
+            Event::CdnClusterCreated(ev) => println!("EVENT {:?}", ev),
+            Event::CdnClusterDistributeRevenues(ev) => println!("EVENT {:?}", ev),
+            Event::CdnNodeCreated(ev) => println!("EVENT {:?}", ev),
         }
     }
 }