diff --git a/contracts/stream_contract/src/lib.rs b/contracts/stream_contract/src/lib.rs
index 62fa2cb..47b77a6 100644
--- a/contracts/stream_contract/src/lib.rs
+++ b/contracts/stream_contract/src/lib.rs
@@ -1,7 +1,18 @@
 #![no_std]
-use soroban_sdk::{contract, contractimpl, contracttype, contracterror, Address, Env, Symbol, symbol_short, token};
 
-#[derive(Clone)]
+use soroban_sdk::{
+    contract, contracterror, contractimpl, contracttype, panic_with_error, token, Address, Env,
+    Symbol,
+};
+
+#[contracttype]
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub enum DataKey {
+    Stream(u64),
+    StreamCounter,
+}
+
+#[derive(Clone, Debug, Eq, PartialEq)]
 #[contracttype]
 pub struct Stream {
     pub sender: Address,
@@ -24,7 +35,6 @@ pub enum StreamError {
     StreamInactive = 4,
 }
 
-// Event definitions for indexing
 #[contracttype]
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub struct StreamCreatedEvent {
@@ -77,26 +87,48 @@ impl StreamContract {
         duration: u64,
     ) -> u64 {
         sender.require_auth();
-        // Placeholder for stream creation logic
-        // 1. Transfer tokens to contract
-        // 2. Store stream state
 
-        // Generate stream ID (placeholder - use proper counter in production)
-        let stream_id: u64 = env.ledger().sequence() as u64;
+        if amount <= 0 || duration == 0 {
+            panic_with_error!(&env, StreamError::InvalidAmount);
+        }
+
+        let stream_id = Self::get_next_stream_id(&env);
         let start_time = env.ledger().timestamp();
+        let rate_per_second = amount / (duration as i128);
+
+        let token_client = token::Client::new(&env, &token_address);
+        let contract_address = env.current_contract_address();
+        token_client.transfer(&sender, &contract_address, &amount);
+
+        let stream = Stream {
+            sender: sender.clone(),
+            recipient: recipient.clone(),
+            token_address: token_address.clone(),
+            rate_per_second,
+            deposited_amount: amount,
+            withdrawn_amount: 0,
+            start_time,
+            last_update_time: start_time,
+            is_active: true,
+        };
+
+        env.storage()
+            .persistent()
+            .set(&DataKey::Stream(stream_id), &stream);
 
-        // Emit StreamCreated event
         env.events().publish(
             (Symbol::new(&env, "stream_created"), stream_id),
             StreamCreatedEvent {
                 stream_id,
-                sender: sender.clone(),
-                recipient: recipient.clone(),
-                rate,
-                token_address: token_address.clone(),
+                sender,
+                recipient,
+                rate: rate_per_second,
+                token_address,
                 start_time,
-            }
+            },
         );
+
+        stream_id
     }
 
     fn get_next_stream_id(env: &Env) -> u64 {
@@ -112,158 +144,144 @@ impl StreamContract {
         next_id
     }
 
-    pub fn withdraw(_env: Env, _recipient: Address, _stream_id: u64) {
-        // Placeholder for withdraw logic
-        // 1. Calculate claimable amount based on time delta
-        // 2. Transfer tokens to recipient
-        // 3. Update stream state
+    pub fn withdraw(env: Env, recipient: Address, stream_id: u64) -> Result<i128, StreamError> {
+        recipient.require_auth();
 
-        // Placeholder amount calculation
-        let amount: i128 = 0; // Calculate actual amount in production
-        let timestamp = env.ledger().timestamp();
+        let storage = env.storage().persistent();
+        let stream_key = DataKey::Stream(stream_id);
 
-        // Emit TokensWithdrawn event
-        env.events().publish(
-            (Symbol::new(&env, "tokens_withdrawn"), stream_id),
-            TokensWithdrawnEvent {
-                stream_id,
-                recipient: recipient.clone(),
-                amount,
-                timestamp,
-            }
-        );
-    }
+        let mut stream: Stream = match storage.get(&stream_key) {
+            Some(s) => s,
+            None => return Err(StreamError::StreamNotFound),
+        };
 
-    pub fn cancel_stream(env: Env, sender: Address, stream_id: u64) {
-        sender.require_auth();
-        // Placeholder for cancel logic
-        // 1. Calculate amount already withdrawn
-        // 2. Return remaining tokens to sender
-        // 3. Mark stream as cancelled
+        if stream.recipient != recipient {
+            return Err(StreamError::Unauthorized);
+        }
 
-        // Placeholder values
-        let recipient = sender.clone(); // Get actual recipient from storage in production
-        let amount_withdrawn: i128 = 0; // Calculate actual amount in production
+        if !stream.is_active {
+            return Err(StreamError::StreamInactive);
+        }
+
+        let claimable = stream.deposited_amount - stream.withdrawn_amount;
+        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();
+        if stream.withdrawn_amount >= stream.deposited_amount {
+            stream.is_active = false;
+        }
+        storage.set(&stream_key, &stream);
 
-        // Emit StreamCancelled event
         env.events().publish(
-            (Symbol::new(&env, "stream_cancelled"), stream_id),
-            StreamCancelledEvent {
+            (Symbol::new(&env, "tokens_withdrawn"), stream_id),
+            TokensWithdrawnEvent {
                 stream_id,
-                sender: sender.clone(),
                 recipient,
-                amount_withdrawn,
-            }
+                amount: claimable,
+                timestamp: stream.last_update_time,
+            },
         );
+
+        Ok(claimable)
     }
 
-    /// Allows the sender to add more funds to an existing stream
-    /// This extends the duration of the stream without creating a new one
-    pub fn top_up_stream(env: Env, sender: Address, stream_id: u64, amount: i128) -> Result<(), StreamError> {
-        // Require sender authentication
+    pub fn cancel_stream(env: Env, sender: Address, stream_id: u64) -> Result<(), StreamError> {
         sender.require_auth();
 
-        // Validate amount is positive
-        if amount <= 0 {
-            return Err(StreamError::InvalidAmount);
-        }
-
-        // Get the stream from storage
         let storage = env.storage().persistent();
-        let stream_key = (symbol_short!("STREAMS"), stream_id);
+        let stream_key = DataKey::Stream(stream_id);
 
         let mut stream: Stream = match storage.get(&stream_key) {
             Some(s) => s,
             None => return Err(StreamError::StreamNotFound),
         };
 
-        // Verify the caller is the original sender
         if stream.sender != sender {
             return Err(StreamError::Unauthorized);
         }
 
-        // Verify stream is still active
         if !stream.is_active {
             return Err(StreamError::StreamInactive);
         }
 
-        // Transfer tokens from sender to contract
-        let token_client = token::Client::new(&env, &stream.token_address);
-        let contract_address = env.current_contract_address();
-        token_client.transfer(&sender, &contract_address, &amount);
-
-        // Update stream state with additional deposit
-        stream.deposited_amount += amount;
+        stream.is_active = false;
         stream.last_update_time = env.ledger().timestamp();
-
-        // Save updated stream back to storage
+        let recipient = stream.recipient.clone();
+        let amount_withdrawn = stream.withdrawn_amount;
         storage.set(&stream_key, &stream);
 
-        // Emit StreamToppedUp event
         env.events().publish(
-            (Symbol::new(&env, "stream_topped_up"), stream_id),
-            StreamToppedUpEvent {
+            (Symbol::new(&env, "stream_cancelled"), stream_id),
+            StreamCancelledEvent {
                 stream_id,
-                sender: sender.clone(),
-                amount,
-                new_deposited_amount: stream.deposited_amount,
-            }
+                sender,
+                recipient,
+                amount_withdrawn,
+            },
         );
 
         Ok(())
     }
 
-    /// Allows the sender to add more funds to an existing stream
-    /// This extends the duration of the stream without creating a new one
-    pub fn top_up_stream(env: Env, sender: Address, stream_id: u64, amount: i128) -> Result<(), StreamError> {
-        // Require sender authentication
+    pub fn top_up_stream(
+        env: Env,
+        sender: Address,
+        stream_id: u64,
+        amount: i128,
+    ) -> Result<(), StreamError> {
         sender.require_auth();
 
-        // Validate amount is positive
         if amount <= 0 {
             return Err(StreamError::InvalidAmount);
         }
 
-        // Get the stream from storage
         let storage = env.storage().persistent();
-        let stream_key = (symbol_short!("STREAMS"), stream_id);
+        let stream_key = DataKey::Stream(stream_id);
 
         let mut stream: Stream = match storage.get(&stream_key) {
             Some(s) => s,
             None => return Err(StreamError::StreamNotFound),
         };
 
-        // Verify the caller is the original sender
         if stream.sender != sender {
             return Err(StreamError::Unauthorized);
         }
 
-        // Verify stream is still active
         if !stream.is_active {
             return Err(StreamError::StreamInactive);
         }
 
-        // Transfer tokens from sender to contract
         let token_client = token::Client::new(&env, &stream.token_address);
         let contract_address = env.current_contract_address();
         token_client.transfer(&sender, &contract_address, &amount);
 
-        // Update stream state with additional deposit
         stream.deposited_amount += amount;
         stream.last_update_time = env.ledger().timestamp();
 
-        // Save updated stream back to storage
         storage.set(&stream_key, &stream);
 
-        Ok(())
-    }
+        env.events().publish(
+            (Symbol::new(&env, "stream_topped_up"), stream_id),
+            StreamToppedUpEvent {
+                stream_id,
+                sender,
+                amount,
+                new_deposited_amount: stream.deposited_amount,
+            },
+        );
 
-    pub fn get_stream(env: Env, stream_id: u64) -> Option<Stream> {
-        env.storage().instance().get(&DataKey::Stream(stream_id))
+        Ok(())
     }
 
     pub fn get_stream(env: Env, stream_id: u64) -> Option<Stream> {
-        env.storage().instance().get(&DataKey::Stream(stream_id))
+        env.storage().persistent().get(&DataKey::Stream(stream_id))
     }
 }
 
diff --git a/contracts/stream_contract/src/test.rs b/contracts/stream_contract/src/test.rs
index 6a11e0e..3d2a824 100644
--- a/contracts/stream_contract/src/test.rs
+++ b/contracts/stream_contract/src/test.rs
@@ -1,10 +1,18 @@
 #![cfg(test)]
 
+extern crate std;
+
 use super::*;
-use soroban_sdk::{Env, testutils::Address as _, Address, token, symbol_short};
+use soroban_sdk::{testutils::Address as _, token, xdr, Address, Env, Symbol, TryFromVal};
+
+fn create_token_contract(env: &Env) -> (Address, Address) {
+    let admin = Address::generate(env);
+    let token = env.register_stellar_asset_contract_v2(admin.clone());
+    (token.address(), admin)
+}
 
 #[test]
-fn test_create_stream() {
+fn test_create_stream_persists_state() {
     let env = Env::default();
     env.mock_all_auths();
 
@@ -12,35 +20,27 @@ fn test_create_stream() {
     let sender = Address::generate(&env);
     let recipient = Address::generate(&env);
 
-    let stellar_asset = soroban_sdk::token::StellarAssetClient::new(&env, &token_address);
-    stellar_asset.mint(&sender, &1000);
+    let stellar_asset = token::StellarAssetClient::new(&env, &token_address);
+    stellar_asset.mint(&sender, &1_000);
 
     let contract_id = env.register(StreamContract, ());
     let client = StreamContractClient::new(&env, &contract_id);
 
-    let token_client = soroban_sdk::token::Client::new(&env, &token_address);
-    token_client.approve(&sender, &contract_id, &500, &1000000);
-
-    let amount: i128 = 500;
-    let duration: u64 = 86400;
-
-    let stream_id = client.create_stream(&sender, &recipient, &token_address, &amount, &duration);
-
+    let stream_id = client.create_stream(&sender, &recipient, &token_address, &500, &100);
     assert_eq!(stream_id, 1);
 
-    let stream = client.get_stream(&stream_id);
-    assert!(stream.is_some());
-    let stream = stream.unwrap();
+    let stream = client.get_stream(&stream_id).unwrap();
     assert_eq!(stream.sender, sender);
     assert_eq!(stream.recipient, recipient);
-    assert_eq!(stream.rate, amount);
     assert_eq!(stream.token_address, token_address);
-    assert_eq!(stream.duration, duration);
-    assert_eq!(stream.withdrawn, 0);
+    assert_eq!(stream.rate_per_second, 5);
+    assert_eq!(stream.deposited_amount, 500);
+    assert_eq!(stream.withdrawn_amount, 0);
+    assert!(stream.is_active);
 }
 
 #[test]
-fn test_create_multiple_streams() {
+fn test_create_multiple_streams_increments_counter() {
     let env = Env::default();
     env.mock_all_auths();
 
@@ -49,109 +49,96 @@ fn test_create_multiple_streams() {
     let recipient1 = Address::generate(&env);
     let recipient2 = Address::generate(&env);
 
-    let stellar_asset = soroban_sdk::token::StellarAssetClient::new(&env, &token_address);
-    stellar_asset.mint(&sender, &2000);
+    let stellar_asset = token::StellarAssetClient::new(&env, &token_address);
+    stellar_asset.mint(&sender, &2_000);
 
     let contract_id = env.register(StreamContract, ());
     let client = StreamContractClient::new(&env, &contract_id);
 
-    let token_client = soroban_sdk::token::Client::new(&env, &token_address);
-    token_client.approve(&sender, &contract_id, &1000, &1000000);
-
-    let stream_id1 = client.create_stream(&sender, &recipient1, &token_address, &500, &86400);
-    let stream_id2 = client.create_stream(&sender, &recipient2, &token_address, &500, &86400);
+    let stream_id1 = client.create_stream(&sender, &recipient1, &token_address, &500, &100);
+    let stream_id2 = client.create_stream(&sender, &recipient2, &token_address, &500, &100);
 
     assert_eq!(stream_id1, 1);
     assert_eq!(stream_id2, 2);
 }
 
 #[test]
-fn test_create_stream_transfers_tokens() {
+fn test_withdraw_rejects_non_recipient() {
     let env = Env::default();
     env.mock_all_auths();
 
     let (token_address, _admin) = create_token_contract(&env);
     let sender = Address::generate(&env);
     let recipient = Address::generate(&env);
+    let attacker = Address::generate(&env);
 
-    let stellar_asset = soroban_sdk::token::StellarAssetClient::new(&env, &token_address);
-    stellar_asset.mint(&sender, &1000);
+    let stellar_asset = token::StellarAssetClient::new(&env, &token_address);
+    stellar_asset.mint(&sender, &1_000);
 
     let contract_id = env.register(StreamContract, ());
     let client = StreamContractClient::new(&env, &contract_id);
-    let token_client = soroban_sdk::token::Client::new(&env, &token_address);
-
-    let initial_sender_balance = token_client.balance(&sender);
-    let initial_contract_balance = token_client.balance(&contract_id);
 
-    token_client.approve(&sender, &contract_id, &500, &1000000);
+    let stream_id = client.create_stream(&sender, &recipient, &token_address, &500, &100);
 
-    let amount: i128 = 500;
-    let duration: u64 = 86400;
+    let unauthorized_result = client.try_withdraw(&attacker, &stream_id);
+    assert_eq!(unauthorized_result, Err(Ok(StreamError::Unauthorized)));
 
-    client.create_stream(&sender, &recipient, &token_address, &amount, &duration);
-
-    assert_eq!(
-        token_client.balance(&sender),
-        initial_sender_balance - amount
-    );
-    assert_eq!(
-        token_client.balance(&contract_id),
-        initial_contract_balance + amount
-    );
+    let stream = client.get_stream(&stream_id).unwrap();
+    assert_eq!(stream.withdrawn_amount, 0);
+    assert!(stream.is_active);
 }
 
 #[test]
-fn test_top_up_stream_success() {
+fn test_withdraw_authorized_recipient_receives_tokens() {
     let env = Env::default();
     env.mock_all_auths();
 
+    let (token_address, _admin) = create_token_contract(&env);
+    let sender = Address::generate(&env);
+    let recipient = Address::generate(&env);
+
+    let stellar_asset = token::StellarAssetClient::new(&env, &token_address);
+    stellar_asset.mint(&sender, &1_000);
+
     let contract_id = env.register(StreamContract, ());
     let client = StreamContractClient::new(&env, &contract_id);
+    let token_client = token::Client::new(&env, &token_address);
+
+    let stream_id = client.create_stream(&sender, &recipient, &token_address, &500, &100);
+    let recipient_balance_before = token_client.balance(&recipient);
+
+    let _ = client.withdraw(&recipient, &stream_id);
+
+    let recipient_balance_after = token_client.balance(&recipient);
+    assert_eq!(recipient_balance_after - recipient_balance_before, 500);
+
+    let stream = client.get_stream(&stream_id).unwrap();
+    assert_eq!(stream.withdrawn_amount, 500);
+    assert!(!stream.is_active);
+}
+
+#[test]
+fn test_top_up_stream_success() {
+    let env = Env::default();
+    env.mock_all_auths();
 
-    // Create mock addresses
+    let (token_address, _admin) = create_token_contract(&env);
     let sender = Address::generate(&env);
     let recipient = Address::generate(&env);
-    let token_admin = Address::generate(&env);
-
-    // Deploy a mock token contract
-    let token_contract = env.register_stellar_asset_contract_v2(token_admin.clone());
-    let token_address = token_contract.address();
-    let token_client = token::StellarAssetClient::new(&env, &token_address);
 
-    // Mint initial tokens to sender
-    token_client.mint(&sender, &1_000_000);
+    let stellar_asset = token::StellarAssetClient::new(&env, &token_address);
+    stellar_asset.mint(&sender, &20_000);
 
-    // Manually create a stream in storage (since create_stream is not fully implemented)
-    let stream = Stream {
-        sender: sender.clone(),
-        recipient: recipient.clone(),
-        token_address: token_address.clone(),
-        rate_per_second: 100,
-        deposited_amount: 10_000,
-        withdrawn_amount: 0,
-        start_time: env.ledger().timestamp(),
-        last_update_time: env.ledger().timestamp(),
-        is_active: true,
-    };
+    let contract_id = env.register(StreamContract, ());
+    let client = StreamContractClient::new(&env, &contract_id);
 
-    let stream_id = 1u64;
-    env.as_contract(&contract_id, || {
-        let storage = env.storage().persistent();
-        storage.set(&(symbol_short!("STREAMS"), stream_id), &stream);
-    });
+    let stream_id = client.create_stream(&sender, &recipient, &token_address, &10_000, &100);
 
-    // Top up the stream with additional funds
-    let top_up_amount = 5_000i128;
-    let result = client.try_top_up_stream(&sender, &stream_id, &top_up_amount);
-    assert!(result.is_ok());
+    let top_up_result = client.try_top_up_stream(&sender, &stream_id, &5_000);
+    assert!(top_up_result.is_ok());
 
-    // Verify the stream was updated
-    env.as_contract(&contract_id, || {
-        let storage = env.storage().persistent();
-        let updated_stream: Stream = storage.get(&(symbol_short!("STREAMS"), stream_id)).unwrap();
-        assert_eq!(updated_stream.deposited_amount, 15_000);
-    });
+    let stream = client.get_stream(&stream_id).unwrap();
+    assert_eq!(stream.deposited_amount, 15_000);
 }
 
 #[test]
@@ -159,19 +146,23 @@ fn test_top_up_stream_invalid_amount() {
     let env = Env::default();
     env.mock_all_auths();
 
+    let (token_address, _admin) = create_token_contract(&env);
+    let sender = Address::generate(&env);
+    let recipient = Address::generate(&env);
+
+    let stellar_asset = token::StellarAssetClient::new(&env, &token_address);
+    stellar_asset.mint(&sender, &20_000);
+
     let contract_id = env.register(StreamContract, ());
     let client = StreamContractClient::new(&env, &contract_id);
 
-    let sender = Address::generate(&env);
-    let stream_id = 1u64;
+    let stream_id = client.create_stream(&sender, &recipient, &token_address, &10_000, &100);
 
-    // Try to top up with negative amount
-    let result = client.try_top_up_stream(&sender, &stream_id, &(-100i128));
-    assert_eq!(result, Err(Ok(StreamError::InvalidAmount)));
+    let negative_result = client.try_top_up_stream(&sender, &stream_id, &-100);
+    assert_eq!(negative_result, Err(Ok(StreamError::InvalidAmount)));
 
-    // Try to top up with zero amount
-    let result = client.try_top_up_stream(&sender, &stream_id, &0i128);
-    assert_eq!(result, Err(Ok(StreamError::InvalidAmount)));
+    let zero_result = client.try_top_up_stream(&sender, &stream_id, &0);
+    assert_eq!(zero_result, Err(Ok(StreamError::InvalidAmount)));
 }
 
 #[test]
@@ -183,9 +174,9 @@ fn test_top_up_stream_not_found() {
     let client = StreamContractClient::new(&env, &contract_id);
 
     let sender = Address::generate(&env);
-    let stream_id = 999u64; // Non-existent stream
+    let stream_id = 999_u64;
 
-    let result = client.try_top_up_stream(&sender, &stream_id, &1_000i128);
+    let result = client.try_top_up_stream(&sender, &stream_id, &1_000);
     assert_eq!(result, Err(Ok(StreamError::StreamNotFound)));
 }
 
@@ -194,38 +185,20 @@ fn test_top_up_stream_unauthorized() {
     let env = Env::default();
     env.mock_all_auths();
 
-    let contract_id = env.register(StreamContract, ());
-    let client = StreamContractClient::new(&env, &contract_id);
-
+    let (token_address, _admin) = create_token_contract(&env);
     let sender = Address::generate(&env);
-    let different_sender = Address::generate(&env);
+    let attacker = Address::generate(&env);
     let recipient = Address::generate(&env);
-    let token_admin = Address::generate(&env);
 
-    let token_contract = env.register_stellar_asset_contract_v2(token_admin.clone());
-    let token_address = token_contract.address();
+    let stellar_asset = token::StellarAssetClient::new(&env, &token_address);
+    stellar_asset.mint(&sender, &20_000);
 
-    // Create a stream with original sender
-    let stream = Stream {
-        sender: sender.clone(),
-        recipient: recipient.clone(),
-        token_address: token_address.clone(),
-        rate_per_second: 100,
-        deposited_amount: 10_000,
-        withdrawn_amount: 0,
-        start_time: env.ledger().timestamp(),
-        last_update_time: env.ledger().timestamp(),
-        is_active: true,
-    };
+    let contract_id = env.register(StreamContract, ());
+    let client = StreamContractClient::new(&env, &contract_id);
 
-    let stream_id = 1u64;
-    env.as_contract(&contract_id, || {
-        let storage = env.storage().persistent();
-        storage.set(&(symbol_short!("STREAMS"), stream_id), &stream);
-    });
+    let stream_id = client.create_stream(&sender, &recipient, &token_address, &10_000, &100);
 
-    // Try to top up with different sender
-    let result = client.try_top_up_stream(&different_sender, &stream_id, &1_000i128);
+    let result = client.try_top_up_stream(&attacker, &stream_id, &1_000);
     assert_eq!(result, Err(Ok(StreamError::Unauthorized)));
 }
 
@@ -234,36 +207,55 @@ fn test_top_up_stream_inactive() {
     let env = Env::default();
     env.mock_all_auths();
 
+    let (token_address, _admin) = create_token_contract(&env);
+    let sender = Address::generate(&env);
+    let recipient = Address::generate(&env);
+
+    let stellar_asset = token::StellarAssetClient::new(&env, &token_address);
+    stellar_asset.mint(&sender, &20_000);
+
     let contract_id = env.register(StreamContract, ());
     let client = StreamContractClient::new(&env, &contract_id);
 
-    let sender = Address::generate(&env);
-    let recipient = Address::generate(&env);
-    let token_admin = Address::generate(&env);
+    let stream_id = client.create_stream(&sender, &recipient, &token_address, &10_000, &100);
+    let _ = client.cancel_stream(&sender, &stream_id);
 
-    let token_contract = env.register_stellar_asset_contract_v2(token_admin.clone());
-    let token_address = token_contract.address();
+    let result = client.try_top_up_stream(&sender, &stream_id, &1_000);
+    assert_eq!(result, Err(Ok(StreamError::StreamInactive)));
+}
+
+#[test]
+fn datakey_stream_serializes_deterministically_and_works_in_storage() {
+    let env = Env::default();
+    let contract_id = env.register(StreamContract, ());
+    let key = DataKey::Stream(42_u64);
+
+    let key_scval_a: xdr::ScVal = (&key).try_into().unwrap();
+    let key_scval_b: xdr::ScVal = (&key).try_into().unwrap();
+    assert_eq!(key_scval_a, key_scval_b);
+
+    let expected_key_scval: xdr::ScVal =
+        (&(Symbol::new(&env, "Stream"), 42_u64)).try_into().unwrap();
+    assert_eq!(key_scval_a, expected_key_scval);
+
+    let decoded_key = DataKey::try_from_val(&env, &key_scval_a).unwrap();
+    assert_eq!(decoded_key, key);
 
-    // Create an inactive stream
     let stream = Stream {
-        sender: sender.clone(),
-        recipient: recipient.clone(),
-        token_address: token_address.clone(),
+        sender: Address::generate(&env),
+        recipient: Address::generate(&env),
+        token_address: Address::generate(&env),
         rate_per_second: 100,
-        deposited_amount: 10_000,
+        deposited_amount: 1_000,
         withdrawn_amount: 0,
-        start_time: env.ledger().timestamp(),
-        last_update_time: env.ledger().timestamp(),
-        is_active: false, // Inactive stream
+        start_time: 1,
+        last_update_time: 1,
+        is_active: true,
     };
 
-    let stream_id = 1u64;
     env.as_contract(&contract_id, || {
-        let storage = env.storage().persistent();
-        storage.set(&(symbol_short!("STREAMS"), stream_id), &stream);
+        env.storage().persistent().set(&key, &stream);
+        let stored: Stream = env.storage().persistent().get(&key).unwrap();
+        assert_eq!(stored, stream);
     });
-
-    // Try to top up inactive stream
-    let result = client.try_top_up_stream(&sender, &stream_id, &1_000i128);
-    assert_eq!(result, Err(Ok(StreamError::StreamInactive)));
 }