fix(contracts): add stream_count view fn and fix resume_stream status…

contracts/stream_contract/src/lib.rs

@@ -31,13 +31,13 @@ pub struct StreamContract;
 
 #[contractimpl]
 impl StreamContract {
-    // ─── Protocol Administration ──────────────────────────────────────────────
+    // ΓöÇΓöÇΓöÇ Protocol Administration ΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇ
 
     /// One-time initialization of the protocol fee configuration.
     ///
     /// # Errors
-    /// - `AlreadyInitialized` — called more than once.
-    /// - `InvalidFeeRate`     — `fee_rate_bps` exceeds `MAX_FEE_RATE_BPS`.
+    /// - `AlreadyInitialized` ΓÇö called more than once.
+    /// - `InvalidFeeRate`     ΓÇö `fee_rate_bps` exceeds `MAX_FEE_RATE_BPS`.
     pub fn initialize(
         env: Env,
         admin: Address,
@@ -77,9 +77,9 @@ impl StreamContract {
     /// Update the treasury address and/or fee rate. Admin-only.
     ///
     /// # Errors
-    /// - `NotInitialized` — `initialize` has not been called.
-    /// - `NotAdmin`       — caller is not the current admin.
-    /// - `InvalidFeeRate` — `fee_rate_bps` exceeds `MAX_FEE_RATE_BPS`.
+    /// - `NotInitialized` ΓÇö `initialize` has not been called.
+    /// - `NotAdmin`       ΓÇö caller is not the current admin.
+    /// - `InvalidFeeRate` ΓÇö `fee_rate_bps` exceeds `MAX_FEE_RATE_BPS`.
     pub fn update_fee_config(
         env: Env,
         admin: Address,
@@ -125,8 +125,8 @@ impl StreamContract {
     /// becomes the sole admin and the previous admin loses all admin privileges.
     ///
     /// # Errors
-    /// - `NotInitialized` — `initialize` has not been called.
-    /// - `NotAdmin`       — caller is not the current admin.
+    /// - `NotInitialized` ΓÇö `initialize` has not been called.
+    /// - `NotAdmin`       ΓÇö caller is not the current admin.
     pub fn transfer_admin(
         env: Env,
         current_admin: Address,
@@ -164,7 +164,7 @@ impl StreamContract {
         try_load_config(&env)
     }
 
-    // ─── Stream Operations ────────────────────────────────────────────────────
+    // ΓöÇΓöÇΓöÇ Stream Operations ΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇ
 
     /// Create a new payment stream.
     ///
@@ -211,7 +211,7 @@ impl StreamContract {
 
         // Reject streams where integer division rounds the rate to zero.
         // Such a stream would lock the sender's tokens in the contract while
-        // never accruing anything to the recipient — almost always a caller
+        // never accruing anything to the recipient ΓÇö almost always a caller
         // mistake (wrong decimals or an excessively long duration).
         // Soroban rolls back the entire transaction on Err, so the token
         // transfer above is unwound automatically.
@@ -260,10 +260,10 @@ impl StreamContract {
     /// is subject to protocol fees (if configured) before being added to the stream.
     ///
     /// # Errors
-    /// - `InvalidAmount`   — `amount` ≤ 0.
-    /// - `StreamNotFound`  — no stream exists with `stream_id`.
-    /// - `Unauthorized`    — caller is not the stream's sender.
-    /// - `StreamInactive`  — stream has been cancelled or fully withdrawn.
+    /// - `InvalidAmount`   ΓÇö `amount` Γëñ 0.
+    /// - `StreamNotFound`  ΓÇö no stream exists with `stream_id`.
+    /// - `Unauthorized`    ΓÇö caller is not the stream's sender.
+    /// - `StreamInactive`  ΓÇö stream has been cancelled or fully withdrawn.
     pub fn top_up_stream(
         env: Env,
         sender: Address,
@@ -311,7 +311,7 @@ impl StreamContract {
         Ok(())
     }
 
-    // ─── Internal Helpers ─────────────────────────────────────────────────────
+    // ΓöÇΓöÇΓöÇ Internal Helpers ΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇ
 
     /// Ensures the supplied token address implements the Soroban token interface.
     fn validate_token_contract(env: &Env, token_address: &Address) -> Result<(), StreamError> {
@@ -365,8 +365,8 @@ impl StreamContract {
     /// Validate that a stream exists and is owned by the caller.
     ///
     /// # Errors
-    /// - `StreamNotFound` — no stream exists with `stream_id`.
-    /// - `Unauthorized` — caller is not the stream's sender.
+    /// - `StreamNotFound` ΓÇö no stream exists with `stream_id`.
+    /// - `Unauthorized` ΓÇö caller is not the stream's sender.
     fn validate_stream_ownership(stream: &Stream, caller: &Address) -> Result<(), StreamError> {
         if stream.sender != *caller {
             return Err(StreamError::Unauthorized);
@@ -377,7 +377,7 @@ impl StreamContract {
     /// Validate that a stream is active.
     ///
     /// # Errors
-    /// - `StreamInactive` — stream has been cancelled or fully withdrawn.
+    /// - `StreamInactive` ΓÇö stream has been cancelled or fully withdrawn.
     fn validate_stream_active(stream: &Stream) -> Result<(), StreamError> {
         if !stream.is_active {
             return Err(StreamError::StreamInactive);
@@ -423,10 +423,10 @@ impl StreamContract {
     /// inactive once fully drained.
     ///
     /// # Errors
-    /// - `StreamNotFound`  — no stream exists with `stream_id`.
-    /// - `Unauthorized`    — caller is not the stream's recipient.
-    /// - `StreamInactive`  — stream is already inactive.
-    /// - `InvalidAmount`   — no claimable balance (fully withdrawn already).
+    /// - `StreamNotFound`  ΓÇö no stream exists with `stream_id`.
+    /// - `Unauthorized`    ΓÇö caller is not the stream's recipient.
+    /// - `StreamInactive`  ΓÇö stream is already inactive.
+    /// - `InvalidAmount`   ΓÇö no claimable balance (fully withdrawn already).
     pub fn withdraw(env: Env, recipient: Address, stream_id: u64) -> Result<i128, StreamError> {
         recipient.require_auth();
 
@@ -487,9 +487,9 @@ impl StreamContract {
     /// balance is refunded to the sender.
     ///
     /// # Errors
-    /// - `StreamNotFound`  — no stream exists with `stream_id`.
-    /// - `Unauthorized`    — caller is not the stream's sender.
-    /// - `StreamInactive`  — stream is already inactive.
+    /// - `StreamNotFound`  ΓÇö no stream exists with `stream_id`.
+    /// - `Unauthorized`    ΓÇö caller is not the stream's sender.
+    /// - `StreamInactive`  ΓÇö stream is already inactive.
     pub fn cancel_stream(env: Env, sender: Address, stream_id: u64) -> Result<(), StreamError> {
         sender.require_auth();
 
@@ -510,9 +510,10 @@ impl StreamContract {
         let refunded_amount = stream
             .deposited_amount
             .saturating_sub(stream.withdrawn_amount);
-
         stream.is_active = false;
         stream.status = StreamStatus::Cancelled;
+        stream.paused = false;
+        stream.paused_at = None;
         stream.last_update_time = now;
 
         let recipient = stream.recipient.clone();
@@ -551,9 +552,9 @@ impl StreamContract {
     /// Pause an active stream. Only the sender may pause.
     ///
     /// # Errors
-    /// - `StreamNotFound`  — no stream exists with `stream_id`.
-    /// - `Unauthorized`    — caller is not the stream's sender.
-    /// - `StreamInactive`  — stream is already inactive.
+    /// - `StreamNotFound`  ΓÇö no stream exists with `stream_id`.
+    /// - `Unauthorized`    ΓÇö caller is not the stream's sender.
+    /// - `StreamInactive`  ΓÇö stream is already inactive.
     pub fn pause_stream(env: Env, sender: Address, stream_id: u64) -> Result<(), StreamError> {
         sender.require_auth();
 
@@ -590,15 +591,21 @@ impl StreamContract {
     /// duration it was paused.
     ///
     /// # Errors
-    /// - `StreamNotFound`  — no stream exists with `stream_id`.
-    /// - `Unauthorized`    — caller is not the stream's sender.
-    /// - `StreamInactive`  — stream is not paused (already active or cancelled).
+    /// - `StreamNotFound`  ΓÇö no stream exists with `stream_id`.
+    /// - `Unauthorized`    ΓÇö caller is not the stream's sender.
+    /// - `StreamInactive`  ΓÇö stream is not paused (already active or cancelled).
     pub fn resume_stream(env: Env, sender: Address, stream_id: u64) -> Result<u64, StreamError> {
         sender.require_auth();
 
         let mut stream = load_stream(&env, stream_id)?;
         Self::validate_stream_ownership(&stream, &sender)?;
 
+        // Reject if the stream is not in Paused status — this covers streams
+        // that were cancelled while paused (is_active=false, paused=true).
+        if stream.status != StreamStatus::Paused {
+            return Err(StreamError::StreamInactive);
+        }
+
         if !stream.paused {
             return Err(StreamError::StreamInactive);
         }
@@ -633,7 +640,21 @@ impl StreamContract {
         Ok(new_end_time)
     }
 
-    // ─── Read-only Queries ────────────────────────────────────────────────────
+    // ΓöÇΓöÇΓöÇ Read-only Queries ΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇ
+
+    /// Returns the total number of streams ever created (monotonically increasing).
+    ///
+    /// This is the global stream ID counter, not the count of currently active
+    /// streams. It equals the highest stream ID that has been assigned, making
+    /// it useful for cursor-based or offset pagination without a full DB scan.
+    ///
+    /// Returns `0` on a freshly-deployed contract where no stream has been created.
+    pub fn stream_count(env: Env) -> u64 {
+        env.storage()
+            .instance()
+            .get(&crate::types::DataKey::StreamCounter)
+            .unwrap_or(0)
+    }
 
     /// Returns the stream record for `stream_id`, or `None` if it does not exist.
     pub fn get_stream(env: Env, stream_id: u64) -> Option<Stream> {
@@ -663,7 +684,7 @@ impl StreamContract {
         })
     }
 
-    // ─── Internal Helpers ─────────────────────────────────────────────────────
+    // ΓöÇΓöÇΓöÇ Internal Helpers ΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇ
 
     /// Deducts the protocol fee from `amount`, transfers it to the treasury,
     /// emits a `fee_collected` event, and returns the net amount.

contracts/stream_contract/src/test.rs

@@ -2530,6 +2530,177 @@ fn test_resume_stream_emits_event() {
     assert_eq!(payload.new_end_time, 1150);
 }
 
+// ─── #421 stream_count ────────────────────────────────────────────────────────
+
+#[test]
+fn test_stream_count_returns_zero_on_fresh_contract() {
+    // A freshly deployed contract with no streams must return 0.
+    let env = Env::default();
+    env.mock_all_auths();
+    let client = create_contract(&env);
+    assert_eq!(client.stream_count(), 0);
+}
+
+#[test]
+fn test_stream_count_increments_by_one_per_create() {
+    let env = Env::default();
+    env.mock_all_auths();
+    let (token, _) = create_token(&env);
+    let sender = Address::generate(&env);
+    mint(&env, &token, &sender, 3_000);
+
+    let client = create_contract(&env);
+
+    assert_eq!(client.stream_count(), 0);
+
+    client.create_stream(&sender, &Address::generate(&env), &token, &1_000, &100);
+    assert_eq!(client.stream_count(), 1);
+
+    client.create_stream(&sender, &Address::generate(&env), &token, &1_000, &100);
+    assert_eq!(client.stream_count(), 2);
+
+    client.create_stream(&sender, &Address::generate(&env), &token, &1_000, &100);
+    assert_eq!(client.stream_count(), 3);
+}
+
+#[test]
+fn test_stream_count_is_not_decremented_by_cancel() {
+    // stream_count counts all streams ever created, not just active ones.
+    let env = Env::default();
+    env.mock_all_auths();
+    let (token, _) = create_token(&env);
+    let sender = Address::generate(&env);
+    mint(&env, &token, &sender, 1_000);
+
+    let client = create_contract(&env);
+    let id = client.create_stream(&sender, &Address::generate(&env), &token, &1_000, &1_000);
+    assert_eq!(client.stream_count(), 1);
+
+    client.cancel_stream(&sender, &id);
+    // Cancelling must NOT decrement the counter.
+    assert_eq!(client.stream_count(), 1);
+}
+
+#[test]
+fn test_stream_count_matches_last_stream_id() {
+    // The counter equals the highest stream ID that has been issued.
+    let env = Env::default();
+    env.mock_all_auths();
+    let (token, _) = create_token(&env);
+    let sender = Address::generate(&env);
+    mint(&env, &token, &sender, 5_000);
+
+    let client = create_contract(&env);
+    for i in 1u64..=5 {
+        let id = client.create_stream(&sender, &Address::generate(&env), &token, &1_000, &100);
+        assert_eq!(id, i);
+        assert_eq!(client.stream_count(), i);
+    }
+}
+
+// ─── #787 resume_stream must reject cancelled-while-paused streams ────────────
+
+#[test]
+fn test_resume_after_cancel_while_paused_returns_stream_inactive() {
+    // Pause a stream, cancel it while paused, then attempt to resume.
+    // resume_stream must return StreamInactive and must NOT emit stream_resumed.
+    let env = Env::default();
+    env.mock_all_auths();
+    let (token, _) = create_token(&env);
+    let sender = Address::generate(&env);
+    let recipient = Address::generate(&env);
+    mint(&env, &token, &sender, 1_000);
+
+    let client = create_contract(&env);
+    // 1_000 tokens / 1_000 s = 1 token/s
+    let id = client.create_stream(&sender, &recipient, &token, &1_000, &1_000);
+
+    // Pause at t=100.
+    env.ledger().with_mut(|l| l.timestamp += 100);
+    client.pause_stream(&sender, &id);
+
+    // Cancel while paused at t=200.
+    env.ledger().with_mut(|l| l.timestamp += 100);
+    client.cancel_stream(&sender, &id);
+
+    // Verify the stream is correctly marked cancelled.
+    let s = client.get_stream(&id).unwrap();
+    assert!(!s.is_active);
+    assert_eq!(s.status, StreamStatus::Cancelled);
+    // cancel_stream must also clear the pause fields.
+    assert!(!s.paused);
+    assert!(s.paused_at.is_none());
+
+    // Attempting to resume must return StreamInactive.
+    let result = client.try_resume_stream(&sender, &id);
+    assert_eq!(result, Err(Ok(StreamError::StreamInactive)));
+
+    // No stream_resumed event must have been emitted.
+    let events = env.events().all();
+    let resumed_event = events.iter().find(|e| {
+        Symbol::try_from_val(&env, &e.1.get(0).unwrap()).unwrap()
+            == Symbol::new(&env, "stream_resumed")
+    });
+    assert!(
+        resumed_event.is_none(),
+        "stream_resumed must not be emitted after cancel"
+    );
+}
+
+#[test]
+fn test_cancel_while_paused_clears_pause_fields() {
+    // After cancel_stream on a paused stream, paused and paused_at must be cleared.
+    let env = Env::default();
+    env.mock_all_auths();
+    let (token, _) = create_token(&env);
+    let sender = Address::generate(&env);
+    let recipient = Address::generate(&env);
+    mint(&env, &token, &sender, 1_000);
+
+    let client = create_contract(&env);
+    let id = client.create_stream(&sender, &recipient, &token, &1_000, &1_000);
+
+    env.ledger().with_mut(|l| l.timestamp += 200);
+    client.pause_stream(&sender, &id);
+
+    // Verify pause state is set before cancel.
+    let before = client.get_stream(&id).unwrap();
+    assert!(before.paused);
+    assert!(before.paused_at.is_some());
+
+    client.cancel_stream(&sender, &id);
+
+    // After cancel, pause state must be cleared.
+    let after = client.get_stream(&id).unwrap();
+    assert!(!after.paused, "paused flag must be false after cancel");
+    assert!(
+        after.paused_at.is_none(),
+        "paused_at must be None after cancel"
+    );
+    assert_eq!(after.status, StreamStatus::Cancelled);
+}
+
+#[test]
+fn test_cancel_normal_stream_also_clears_pause_fields() {
+    // Cancelling a non-paused stream should set paused=false and paused_at=None
+    // (they are already in that state, but the assignment must be idempotent).
+    let env = Env::default();
+    env.mock_all_auths();
+    let (token, _) = create_token(&env);
+    let sender = Address::generate(&env);
+    mint(&env, &token, &sender, 1_000);
+
+    let client = create_contract(&env);
+    let id = client.create_stream(&sender, &Address::generate(&env), &token, &1_000, &1_000);
+
+    client.cancel_stream(&sender, &id);
+
+    let s = client.get_stream(&id).unwrap();
+    assert!(!s.paused);
+    assert!(s.paused_at.is_none());
+    assert_eq!(s.status, StreamStatus::Cancelled);
+}
+
 // ─── CEI / reentrancy regression (#789) ──────────────────────────────────────
 
 /// Verify that stream state is committed to storage before the token transfer,