Implement threshold with tests (#27)

src/contract.rs

@@ -1,14 +1,16 @@
+use crate::killswitch::execute_cancel_stream_with_threshold;
 use crate::msg::{
     AveragePriceResponse, ConfigResponse, ExecuteMsg, InstantiateMsg, LatestStreamedPriceResponse,
     MigrateMsg, PositionResponse, PositionsResponse, QueryMsg, StreamResponse, StreamsResponse,
     SudoMsg,
 };
 use crate::state::{next_stream_id, Config, Position, Status, Stream, CONFIG, POSITIONS, STREAMS};
+use crate::threshold::ThresholdState;
 use crate::{killswitch, ContractError};
 use cosmwasm_std::{
     attr, entry_point, to_json_binary, Addr, BankMsg, Binary, Coin, CosmosMsg, Decimal, Decimal256,
-    Deps, DepsMut, Env, Fraction, MessageInfo, Order, Response, StdResult, Timestamp, Uint128,
-    Uint256, Uint64,
+    Deps, DepsMut, Env, Fraction, MessageInfo, Order, Response, StdError, StdResult, Timestamp,
+    Uint128, Uint256, Uint64,
 };
 use cw2::{get_contract_version, set_contract_version};
 use semver::Version;
@@ -83,9 +85,10 @@ pub fn execute(
             out_supply,
             start_time,
             end_time,
+            threshold,
         } => execute_create_stream(
             deps, env, info, treasury, name, url, in_denom, out_denom, out_supply, start_time,
-            end_time,
+            end_time, threshold,
         ),
         ExecuteMsg::UpdateOperator {
             stream_id,
@@ -96,6 +99,9 @@ pub fn execute(
             operator_target,
         } => execute_update_position(deps, env, info, stream_id, operator_target),
         ExecuteMsg::UpdateStream { stream_id } => execute_update_stream(deps, env, stream_id),
+        ExecuteMsg::CancelStreamWithThreshold { stream_id } => {
+            execute_cancel_stream_with_threshold(deps, env, info, stream_id)
+        }
         ExecuteMsg::Subscribe {
             stream_id,
             operator_target,
@@ -217,6 +223,7 @@ pub fn execute_create_stream(
     out_supply: Uint128,
     start_time: Timestamp,
     end_time: Timestamp,
+    threshold: Option<Uint128>,
 ) -> Result<Response, ContractError> {
     let config = CONFIG.load(deps.storage)?;
     if end_time < start_time {
@@ -307,6 +314,9 @@ pub fn execute_create_stream(
     let id = next_stream_id(deps.storage)?;
     STREAMS.save(deps.storage, id, &stream)?;
 
+    let threshold_state = ThresholdState::new();
+    threshold_state.set_threshold_if_any(threshold, id, deps.storage)?;
+
     let attr = vec![
         attr("action", "create_stream"),
         attr("id", id.to_string()),
@@ -845,6 +855,11 @@ pub fn execute_finalize_stream(
     if stream.status == Status::Active {
         stream.status = Status::Finalized
     }
+    // If threshold is set and not reached, finalize will fail
+    // Creator should execute cancel_stream_with_threshold to cancel the stream
+    // Only returns error if threshold is set and not reached
+    let thresholds_state = ThresholdState::new();
+    thresholds_state.error_if_not_reached(stream_id, deps.storage, &stream)?;
     STREAMS.save(deps.storage, stream_id, &stream)?;
 
     let config = CONFIG.load(deps.storage)?;
@@ -939,6 +954,9 @@ pub fn execute_exit_stream(
     if stream.last_updated < stream.end_time {
         update_stream(env.block.time, &mut stream)?;
     }
+    let threshold_state = ThresholdState::new();
+
+    threshold_state.error_if_not_reached(stream_id, deps.storage, &stream)?;
     let operator_target =
         maybe_addr(deps.api, operator_target)?.unwrap_or_else(|| info.sender.clone());
     let mut position = POSITIONS.load(deps.storage, (stream_id, &operator_target))?;
@@ -1121,6 +1139,9 @@ pub fn query(deps: Deps, env: Env, msg: QueryMsg) -> StdResult<Binary> {
         QueryMsg::LastStreamedPrice { stream_id } => {
             to_json_binary(&query_last_streamed_price(deps, env, stream_id)?)
         }
+        QueryMsg::Threshold { stream_id } => {
+            to_json_binary(&query_threshold_state(deps, env, stream_id)?)
+        }
     }
 }
 pub fn query_config(deps: Deps) -> StdResult<ConfigResponse> {
@@ -1286,3 +1307,13 @@ pub fn query_last_streamed_price(
         current_streamed_price: stream.current_streamed_price,
     })
 }
+
+pub fn query_threshold_state(
+    deps: Deps,
+    _env: Env,
+    stream_id: u64,
+) -> Result<Option<Uint128>, StdError> {
+    let threshold_state = ThresholdState::new();
+    let threshold = threshold_state.get_threshold(stream_id, deps.storage)?;
+    Ok(threshold)
+}

src/error.rs

@@ -1,3 +1,4 @@
+use crate::threshold::ThresholdError;
 use cosmwasm_std::{ConversionOverflowError, DivideByZeroError, OverflowError, StdError, Uint128};
 use cw_utils::PaymentError;
 use std::convert::Infallible;
@@ -20,6 +21,9 @@ pub enum ContractError {
     #[error("{0}")]
     DivideByZeroError(#[from] DivideByZeroError),
 
+    #[error("{0}")]
+    ThresholdError(#[from] ThresholdError),
+
     #[error("{0}")]
     ConversionOverflowError(#[from] ConversionOverflowError),
 

src/killswitch.rs

@@ -1,5 +1,6 @@
 use crate::contract::{update_position, update_stream};
 use crate::state::{Status, Stream, CONFIG, POSITIONS, STREAMS};
+use crate::threshold::{ThresholdError, ThresholdState};
 use crate::ContractError;
 use cosmwasm_std::{
     attr, BankMsg, Coin, CosmosMsg, DepsMut, Env, MessageInfo, Response, StdResult, Timestamp,
@@ -91,15 +92,38 @@ pub fn execute_withdraw_paused(
 
 pub fn execute_exit_cancelled(
     deps: DepsMut,
-    _env: Env,
+    env: Env,
     info: MessageInfo,
     stream_id: u64,
     operator_target: Option<String>,
 ) -> Result<Response, ContractError> {
-    let stream = STREAMS.load(deps.storage, stream_id)?;
+    let mut stream = STREAMS.load(deps.storage, stream_id)?;
+
+    // This execution requires the stream to be cancelled or
+    // the stream to be ended and the threshold not reached.
+
     // check if stream is cancelled
     if !stream.is_cancelled() {
-        return Err(ContractError::StreamNotCancelled {});
+        let threshold_state = ThresholdState::new();
+        // Threshold should be set
+        let is_set = threshold_state.check_if_threshold_set(stream_id, deps.storage)?;
+        if !is_set {
+            return Err(ContractError::StreamNotCancelled {});
+        }
+
+        // Stream should not be paused
+        // If stream paused now_block can exceed end_block
+        // Stream being appeared as ended only happens when its paused or cancelled
+        if stream.is_paused() == true {
+            return Err(ContractError::StreamNotCancelled {});
+        }
+        // Stream should be ended
+        if stream.end_time > env.block.time {
+            return Err(ContractError::StreamNotCancelled {});
+        }
+        // Update stream before checking threshold
+        update_stream(env.block.time, &mut stream)?;
+        threshold_state.error_if_reached(stream_id, deps.storage, &stream)?;
     }
 
     let operator_target =
@@ -265,6 +289,65 @@ pub fn execute_cancel_stream(
         .add_attribute("status", "cancelled"))
 }
 
+pub fn execute_cancel_stream_with_threshold(
+    deps: DepsMut,
+    env: Env,
+    info: MessageInfo,
+    stream_id: u64,
+) -> Result<Response, ContractError> {
+    let mut stream = STREAMS.load(deps.storage, stream_id)?;
+
+    if env.block.time < stream.end_time {
+        return Err(ContractError::StreamNotEnded {});
+    }
+    if info.sender != stream.treasury {
+        return Err(ContractError::Unauthorized {});
+    }
+
+    // Stream should not be paused or cancelled
+    if stream.is_killswitch_active() {
+        return Err(ContractError::StreamKillswitchActive {});
+    }
+
+    // This should be impossible because creator can not finalize stream when threshold is not reached
+    if stream.status == Status::Finalized {
+        return Err(ContractError::StreamAlreadyFinalized {});
+    }
+
+    if stream.last_updated < stream.end_time {
+        update_stream(env.block.time, &mut stream)?;
+    }
+
+    let threshold_state = ThresholdState::new();
+
+    if !threshold_state.check_if_threshold_set(stream_id, deps.storage)? {
+        return Err(ContractError::ThresholdError(
+            ThresholdError::ThresholdNotSet {},
+        ));
+    }
+    // Threshold should not be reached
+    threshold_state.error_if_reached(stream_id, deps.storage, &stream)?;
+
+    stream.status = Status::Cancelled;
+
+    STREAMS.save(deps.storage, stream_id, &stream)?;
+
+    //Refund all out tokens to stream creator(treasury)
+    let messages: Vec<CosmosMsg> = vec![CosmosMsg::Bank(BankMsg::Send {
+        to_address: stream.treasury.to_string(),
+        amount: vec![Coin {
+            denom: stream.out_denom,
+            amount: stream.out_supply,
+        }],
+    })];
+
+    Ok(Response::new()
+        .add_attribute("action", "cancel_stream")
+        .add_messages(messages)
+        .add_attribute("stream_id", stream_id.to_string())
+        .add_attribute("status", "cancelled"))
+}
+
 pub fn sudo_pause_stream(
     deps: DepsMut,
     env: Env,

src/lib.rs

@@ -9,3 +9,4 @@ pub mod msg;
 pub mod state;
 #[cfg(test)]
 mod tests;
+pub mod threshold;

src/msg.rs

@@ -45,6 +45,8 @@ pub enum ExecuteMsg {
         start_time: Timestamp,
         /// Unix timestamp when the stream ends. Calculations in nano sec precision.
         end_time: Timestamp,
+        /// Minimum amount of `spent_in` for a stream to be finalized.
+        threshold: Option<Uint128>,
     },
     /// Update stream and calculates distribution state.
     UpdateStream {
@@ -119,6 +121,9 @@ pub enum ExecuteMsg {
         /// operator_target is the address of operator targets to execute on behalf of the user.
         operator_target: Option<String>,
     },
+    CancelStreamWithThreshold {
+        stream_id: u64,
+    },
 
     UpdateConfig {
         min_stream_duration: Option<Uint64>,
@@ -168,6 +173,8 @@ pub enum QueryMsg {
     /// Returns currently streaming price of a sale.
     #[returns(LatestStreamedPriceResponse)]
     LastStreamedPrice { stream_id: u64 },
+    #[returns(Uint128)]
+    Threshold { stream_id: u64 },
 }
 
 #[cw_serde]
@@ -193,7 +200,7 @@ pub struct ConfigResponse {
 #[cw_serde]
 pub struct StreamResponse {
     pub id: u64,
-    /// address of the treasury where the stream earnings will be sent.
+    /// Address of the treasury where the stream earnings will be sent.
     pub treasury: String,
     /// URL of the stream.
     pub url: Option<String>,