feat: soroban stream lifecycle implementation

contracts/stream_contract/src/lib.rs

@@ -51,7 +51,14 @@ impl StreamContract {
             return Err(StreamError::InvalidFeeRate);
         }
 
-        save_config(&env, &ProtocolConfig { admin, treasury, fee_rate_bps });
+        save_config(
+            &env,
+            &ProtocolConfig {
+                admin,
+                treasury,
+                fee_rate_bps,
+            },
+        );
         Ok(())
     }
 
@@ -221,6 +228,26 @@ impl StreamContract {
         Ok(())
     }
 
+    // ─── Internal Helpers ─────────────────────────────────────────────────────
+
+    fn calculate_claimable(stream: &Stream, now: u64) -> i128 {
+        let elapsed = now.saturating_sub(stream.last_update_time);
+
+        let streamed = (elapsed as i128)
+            .checked_mul(stream.rate_per_second)
+            .unwrap_or(i128::MAX);
+
+        let remaining = stream
+            .deposited_amount
+            .saturating_sub(stream.withdrawn_amount);
+
+        if streamed > remaining {
+            remaining
+        } else {
+            streamed
+        }
+    }
+
     /// Withdraw all currently claimable tokens from a stream.
     ///
     /// Only the stream's recipient may call this. The stream is marked
@@ -231,30 +258,6 @@ impl StreamContract {
     /// - `Unauthorized`    — caller is not the stream's recipient.
     /// - `StreamInactive`  — stream is already inactive.
     /// - `InvalidAmount`   — no claimable balance (fully withdrawn already).
-
-
-fn calculate_claimable(stream: &Stream, now: u64) -> i128 {
-    let elapsed = now.saturating_sub(stream.last_update_time);
-
-    let streamed = match (elapsed as i128).checked_mul(stream.rate_per_second) {
-        Some(val) => val,
-        None => i128::MAX,
-    };
-
-    let remaining = stream
-        .deposited_amount
-        .saturating_sub(stream.withdrawn_amount);
-
-    if streamed > remaining {
-        remaining
-    } else {
-        streamed
-    }
-}
-
-
-
-
     pub fn withdraw(env: Env, recipient: Address, stream_id: u64) -> Result<i128, StreamError> {
         recipient.require_auth();
 
@@ -270,15 +273,18 @@ fn calculate_claimable(stream: &Stream, now: u64) -> i128 {
         let now = env.ledger().timestamp();
         let claimable = Self::calculate_claimable(&stream, now);
 
-if claimable <= 0 {
-    return Err(StreamError::InvalidAmount);
-}
+        if claimable <= 0 {
+            return Err(StreamError::InvalidAmount);
+        }
+
         let token_client = token::Client::new(&env, &stream.token_address);
         let contract_address = env.current_contract_address();
         token_client.transfer(&contract_address, &recipient, &claimable);
 
         stream.withdrawn_amount += claimable;
-        stream.last_update_time = env.ledger().timestamp();
+        stream.last_update_time = now;
+
+        // Mark stream as inactive if all funds have been withdrawn
         if stream.withdrawn_amount >= stream.deposited_amount {
             stream.is_active = false;
         }
@@ -319,16 +325,24 @@ if claimable <= 0 {
             return Err(StreamError::StreamInactive);
         }
 
-        // Refund the unspent balance to the sender.
-        let refunded_amount = stream.deposited_amount - stream.withdrawn_amount;
+        // Calculate accrued tokens that belong to the recipient
+        let now = env.ledger().timestamp();
+        let accrued_amount = Self::calculate_claimable(&stream, now);
+
+        // Refund only the unspent balance minus accrued tokens (accrued tokens stay for recipient)
+        let refunded_amount = stream
+            .deposited_amount
+            .saturating_sub(stream.withdrawn_amount)
+            .saturating_sub(accrued_amount);
+
         if refunded_amount > 0 {
             let token_client = token::Client::new(&env, &stream.token_address);
             let contract_address = env.current_contract_address();
             token_client.transfer(&contract_address, &sender, &refunded_amount);
         }
 
         stream.is_active = false;
-        stream.last_update_time = env.ledger().timestamp();
+        stream.last_update_time = now;
 
         let recipient = stream.recipient.clone();
         let amount_withdrawn = stream.withdrawn_amount;
@@ -369,11 +383,7 @@ if claimable <= 0 {
                 let fee = amount * (cfg.fee_rate_bps as i128) / 10_000;
                 if fee > 0 {
                     let token_client = token::Client::new(env, token_address);
-                    token_client.transfer(
-                        &env.current_contract_address(),
-                        &cfg.treasury,
-                        &fee,
-                    );
+                    token_client.transfer(&env.current_contract_address(), &cfg.treasury, &fee);
                     env.events().publish(
                         (Symbol::new(env, "fee_collected"), stream_id),
                         FeeCollectedEvent {

contracts/stream_contract/src/storage.rs

@@ -16,9 +16,7 @@ pub fn next_stream_id(env: &Env) -> u64 {
         .get(&DataKey::StreamCounter)
         .unwrap_or(0)
         + 1;
-    env.storage()
-        .instance()
-        .set(&DataKey::StreamCounter, &id);
+    env.storage().instance().set(&DataKey::StreamCounter, &id);
     id
 }
 
@@ -47,18 +45,14 @@ pub fn save_stream(env: &Env, stream_id: u64, stream: &Stream) {
 
 /// Returns the stream if it exists, `None` otherwise (used by read-only queries).
 pub fn try_load_stream(env: &Env, stream_id: u64) -> Option<Stream> {
-    env.storage()
-        .persistent()
-        .get(&DataKey::Stream(stream_id))
+    env.storage().persistent().get(&DataKey::Stream(stream_id))
 }
 
 // ─── Protocol Config ──────────────────────────────────────────────────────────
 
 /// Checks whether the protocol config has already been initialized.
 pub fn config_exists(env: &Env) -> bool {
-    env.storage()
-        .instance()
-        .has(&DataKey::ProtocolConfig)
+    env.storage().instance().has(&DataKey::ProtocolConfig)
 }
 
 /// Loads the protocol config.
@@ -81,7 +75,5 @@ pub fn save_config(env: &Env, config: &ProtocolConfig) {
 /// Reads the protocol config as an `Option` (returns `None` if unset).
 /// Used by optional fee-collection logic.
 pub fn try_load_config(env: &Env) -> Option<ProtocolConfig> {
-    env.storage()
-        .instance()
-        .get(&DataKey::ProtocolConfig)
+    env.storage().instance().get(&DataKey::ProtocolConfig)
 }