fix(streaming): resolve panics, add pause guards, and delegate events

contracts/streaming/src/lib.rs

@@ -367,6 +367,28 @@ pub struct UnpauseEvent {
     pub timestamp: u64,
 }
 
+#[soroban_sdk::contractevent]
+pub struct DelegateSetEvent {
+    pub stream_id: u64,
+    pub recipient: Address,
+    pub delegate: Address,
+    pub timestamp: u64,
+}
+
+#[soroban_sdk::contractevent]
+pub struct DelegateRemovedEvent {
+    pub stream_id: u64,
+    pub recipient: Address,
+    pub timestamp: u64,
+}
+
+#[soroban_sdk::contractevent]
+pub struct StreamCleanedUpEvent {
+    pub stream_id: u64,
+    pub caller: Address,
+    pub timestamp: u64,
+}
+
 // ─── Contract ────────────────────────────────────────────────────────────────
 
 #[contract]
@@ -1257,6 +1279,22 @@ impl StreamingContract {
             .persistent()
             .get(&DataKey::ArchiveSentBy(address))
             .unwrap_or(Vec::new(&env));
+        let len = all.len();
+        let start = core::cmp::min(offset, len);
+        let end = if let Some(limit_end) = offset.checked_add(limit) {
+            core::cmp::min(limit_end, len)
+        } else {
+            len
+        };
+        let mut result = Vec::new(&env);
+        let mut i = start;
+        while i < end {
+            if let Some(id) = all.get(i) {
+                result.push_back(id);
+            }
+            i += 1;
+        }
+        result
         Self::paginate(&env, &all, offset, limit)
     }
 
@@ -1272,6 +1310,22 @@ impl StreamingContract {
             .persistent()
             .get(&DataKey::ArchiveReceivedBy(address))
             .unwrap_or(Vec::new(&env));
+        let len = all.len();
+        let start = core::cmp::min(offset, len);
+        let end = if let Some(limit_end) = offset.checked_add(limit) {
+            core::cmp::min(limit_end, len)
+        } else {
+            len
+        };
+        let mut result = Vec::new(&env);
+        let mut i = start;
+        while i < end {
+            if let Some(id) = all.get(i) {
+                result.push_back(id);
+            }
+            i += 1;
+        }
+        result
         Self::paginate(&env, &all, offset, limit)
     }
 
@@ -1329,6 +1383,13 @@ impl StreamingContract {
             .persistent()
             .remove(&DataKey::Delegate(stream_id));
 
+        StreamCleanedUpEvent {
+            stream_id,
+            caller,
+            timestamp: env.ledger().timestamp(),
+        }
+        .publish(&env);
+
         Ok(())
     }
 
@@ -1359,6 +1420,7 @@ impl StreamingContract {
     ) -> Result<(), StreamError> {
         let stream = Self::load_stream(&env, stream_id)?;
         stream.sender.require_auth();
+        Self::require_not_paused(&env)?;
 
         env.storage()
             .persistent()
@@ -1394,6 +1456,7 @@ impl StreamingContract {
     pub fn set_delegate(env: Env, stream_id: u64, delegate: Address) -> Result<(), StreamError> {
         let stream = Self::load_stream(&env, stream_id)?;
         stream.recipient.require_auth();
+        Self::require_not_paused(&env)?;
 
         env.storage()
             .persistent()
@@ -1404,18 +1467,34 @@ impl StreamingContract {
             PERSISTENT_TTL_LEDGERS,
         );
 
+        DelegateSetEvent {
+            stream_id,
+            recipient: stream.recipient,
+            delegate,
+            timestamp: env.ledger().timestamp(),
+        }
+        .publish(&env);
+
         Ok(())
     }
 
     /// Remove the delegate for a stream.
     pub fn remove_delegate(env: Env, stream_id: u64) -> Result<(), StreamError> {
         let stream = Self::load_stream(&env, stream_id)?;
         stream.recipient.require_auth();
+        Self::require_not_paused(&env)?;
 
         env.storage()
             .persistent()
             .remove(&DataKey::Delegate(stream_id));
 
+        DelegateRemovedEvent {
+            stream_id,
+            recipient: stream.recipient,
+            timestamp: env.ledger().timestamp(),
+        }
+        .publish(&env);
+
         Ok(())
     }
 

contracts/streaming/src/test_features.rs

@@ -6,6 +6,7 @@ use super::*;
 use soroban_sdk::{
     testutils::{Address as _, Ledger},
     token::{Client as TokenClient, StellarAssetClient},
+    vec, Address, Env,
     Address, Env,
 };
 
@@ -269,6 +270,40 @@ fn test_pause_blocks_cancel() {
     client.cancel(&stream_id);
 }
 
+#[test]
+#[should_panic(expected = "Error(Contract, #16)")]
+fn test_pause_blocks_update_stream_metadata() {
+    let t = TestEnv::setup();
+    let now = 1_000_000u64;
+    t.set_time(now);
+
+    let client = t.client();
+    let params = t.default_params(now);
+    let total = params.total_amount;
+
+    client.initialize(&t.sender);
+
+    t.token().approve(
+        &t.sender,
+        &t.contract_id,
+        &total,
+        &(t.env.ledger().sequence() + 500),
+    );
+    let stream_id = client.create_stream(&t.sender, &params);
+
+    client.pause();
+
+    // Attempting to mutate metadata while paused must be rejected with
+    // ContractPaused (#16) — the same guard that protects every other
+    // state-changing function.
+    let metadata = StreamMetadata {
+        name: soroban_sdk::String::from_str(&t.env, "Salary"),
+        category: soroban_sdk::String::from_str(&t.env, "payroll"),
+        memo: soroban_sdk::String::from_str(&t.env, "monthly"),
+    };
+    client.update_stream_metadata(&stream_id, &metadata);
+}
+
 #[test]
 fn test_read_operations_work_while_paused() {
     let t = TestEnv::setup();
@@ -337,6 +372,9 @@ fn test_only_admin_can_pause() {
     let client = t.client();
     client.initialize(&t.sender);
 
+    let other = Address::generate(&t.env);
+
+    // Non-admin should not be able to pause
     // This test assumes mock_all_auths is set; otherwise auth will fail
     client.pause();
 }
@@ -357,6 +395,7 @@ fn test_only_admin_can_unpause() {
     // Only admin can unpause
     client.unpause();
 
+    let stream = client.get_stream(&1u64);
     // Operations should work again now that the contract is unpaused.
     t.token().approve(
         &t.sender,