diff --git a/contracts/stream_contract/src/lib.rs b/contracts/stream_contract/src/lib.rs index 8ac753d1..226ee746 100644 --- a/contracts/stream_contract/src/lib.rs +++ b/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 { 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 { 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 { @@ -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. diff --git a/contracts/stream_contract/src/test.rs b/contracts/stream_contract/src/test.rs index db247615..fbf054bb 100644 --- a/contracts/stream_contract/src/test.rs +++ b/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, diff --git a/frontend/src/lib/soroban.ts b/frontend/src/lib/soroban.ts index 742f4e39..a7a5667e 100644 --- a/frontend/src/lib/soroban.ts +++ b/frontend/src/lib/soroban.ts @@ -360,3 +360,56 @@ export async function resumeStream( nativeToScVal(params.streamId, { type: "u64" }), ]); } + +/** + * Read-only call to the contract's `stream_count` view function. + * + * Returns the total number of streams ever created (monotonically increasing + * stream ID counter). This is NOT the count of currently active streams — + * cancelled and completed streams are still counted. + * + * Returns `0n` on a freshly-deployed contract where no stream has been created. + * Does not require wallet authentication. + */ +export async function fetchStreamCount(): Promise { + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const sdk: any = await import("@stellar/stellar-sdk"); + const { Contract, TransactionBuilder, BASE_FEE, scValToNative, Keypair } = sdk; + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const rpc: any = sdk.rpc ?? sdk.SorobanRpc; + + const server = new rpc.Server(SOROBAN_RPC_URL, { allowHttp: false }); + // stream_count is a read-only view — use a throwaway keypair for simulation. + const throwawayKeypair = Keypair.random(); + const account = await server.getAccount(throwawayKeypair.publicKey()).catch(() => { + // If the throwaway account doesn't exist on-chain, build a minimal account object. + return { accountId: () => throwawayKeypair.publicKey(), sequenceNumber: () => "0", incrementSequenceNumber: () => {} }; + }); + + const contract = new Contract(CONTRACT_ID); + const tx = new TransactionBuilder(account, { + fee: BASE_FEE, + networkPassphrase: NETWORK_PASSPHRASE, + }) + .addOperation(contract.call("stream_count")) + .setTimeout(30) + .build(); + + const simResult = await server.simulateTransaction(tx); + if (rpc.Api?.isSimulationError?.(simResult) ?? simResult?.error) { + throw new SorobanCallError(`stream_count simulation failed: ${simResult.error}`, "NetworkError"); + } + + const rawResult = simResult?.result?.retval; + if (!rawResult) { + // Contract not yet initialized (no streams ever created). + return 0n; + } + + const nativeValue = scValToNative(rawResult); + if (typeof nativeValue === "bigint") return nativeValue; + if (typeof nativeValue === "number") return BigInt(Math.trunc(nativeValue)); + if (typeof nativeValue === "string") return BigInt(nativeValue); + + throw new SorobanCallError("stream_count returned an unexpected value type.", "Unknown"); +}