Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Anthony-19 wants to merge 2 commits into
base: main
Choose a base branch
from
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6708d3e
fix(contracts): add stream_count view fn and fix resume_stream status…
Anthony-19 Jun 30, 2026
eaf5796
fix(contracts): move pause field reset before save_stream in cancel_s…
Anthony-19 Jul 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
137 changes: 74 additions & 63 deletions contracts/stream_contract/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
#![no_std]
#![no_std]

mod errors;
mod events;
Expand Down Expand Up @@ -30,13 +30,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,
Expand Down Expand Up @@ -76,9 +76,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,
Expand Down Expand Up @@ -124,8 +124,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,
Expand Down Expand Up @@ -163,7 +163,7 @@ impl StreamContract {
try_load_config(&env)
}

// ─── Stream Operations ────────────────────────────────────────────────────
// ΓöÇΓöÇΓöÇ Stream Operations ΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇ

/// Create a new payment stream.
///
Expand All @@ -174,9 +174,9 @@ impl StreamContract {
/// Returns the new stream ID (starts at 1, increments monotonically).
///
/// # Errors
/// - `InvalidAmount` — `amount` ≤ 0.
/// - `InvalidDuration` — `duration` is 0.
/// - `InvalidTokenAddress` — `token_address` is not a token contract.
/// - `InvalidAmount` ΓÇö `amount` Γëñ 0.
/// - `InvalidDuration` ΓÇö `duration` is 0.
/// - `InvalidTokenAddress` ΓÇö `token_address` is not a token contract.
pub fn create_stream(
env: Env,
sender: Address,
Expand Down Expand Up @@ -209,7 +209,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.
Expand Down Expand Up @@ -258,10 +258,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,
Expand All @@ -288,9 +288,10 @@ impl StreamContract {
// Collect protocol fee and get net amount
let net_amount = Self::collect_fee(&env, &stream.token_address, amount, stream_id);

// Update stream state
// Update stream state. `last_update_time` is intentionally left untouched:
// it is the accrual anchor for `calculate_claimable`, and advancing it to
// `now` would discard any already-vested, unwithdrawn tokens.
stream.deposited_amount += net_amount;
stream.last_update_time = env.ledger().timestamp();

save_stream(&env, stream_id, &stream);

Expand All @@ -308,7 +309,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> {
Expand Down Expand Up @@ -362,8 +363,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);
Expand All @@ -374,37 +375,43 @@ 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);
}
Ok(())
}

/// Transfer tokens from contract to recipient and update stream state.
/// Apply a withdrawal: update stream state, persist it, then transfer tokens.
///
/// This helper consolidates the token transfer logic and stream state updates
/// to reduce code duplication across withdrawal operations.
fn transfer_and_update_stream(
/// Follows the Checks-Effects-Interactions (CEI) pattern: all state mutations
/// and the storage write complete before the external token transfer fires.
/// A re-entrant call via a malicious token hook therefore sees the already-updated
/// withdrawn_amount in storage and cannot trigger a double payout.
fn apply_withdrawal(
env: &Env,
stream: &mut Stream,
stream_id: u64,
recipient: &Address,
amount: i128,
now: u64,
) {
let token_client = token::Client::new(env, &stream.token_address);
let contract_address = env.current_contract_address();
token_client.transfer(&contract_address, recipient, &amount);

// Effects: update stream state
stream.withdrawn_amount += amount;
stream.last_update_time = now;

// Mark stream as inactive and completed if fully drained
if stream.withdrawn_amount >= stream.deposited_amount {
stream.is_active = false;
stream.status = StreamStatus::Completed;
}

// Persist state before any external call (CEI)
save_stream(env, stream_id, stream);

// Interaction: transfer tokens only after state is committed to storage
let token_client = token::Client::new(env, &stream.token_address);
token_client.transfer(&env.current_contract_address(), recipient, &amount);
}

/// Withdraw all currently claimable tokens from a stream.
Expand All @@ -414,10 +421,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();

Expand All @@ -431,7 +438,7 @@ impl StreamContract {
// Validate stream is active and not paused
Self::validate_stream_active(&stream)?;
if stream.paused {
return Err(StreamError::StreamInactive);
return Err(StreamError::StreamPaused);
}

let now = env.ledger().timestamp();
Expand All @@ -441,11 +448,10 @@ impl StreamContract {
return Err(StreamError::InvalidAmount);
}

// Use helper function to transfer tokens and update state
Self::transfer_and_update_stream(&env, &mut stream, &recipient, claimable, now);
// Apply withdrawal: updates state, persists to storage, then transfers (CEI)
Self::apply_withdrawal(&env, &mut stream, stream_id, &recipient, claimable, now);

let completed = stream.status == StreamStatus::Completed;
save_stream(&env, stream_id, &stream);

env.events().publish(
(Symbol::new(&env, "tokens_withdrawn"), stream_id),
Expand Down Expand Up @@ -479,9 +485,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();

Expand All @@ -494,34 +500,39 @@ impl StreamContract {
let now = env.ledger().timestamp();
let accrued_amount = Self::calculate_claimable(&stream, now);

let token_client = token::Client::new(&env, &stream.token_address);
let contract_address = env.current_contract_address();

// Settle recipient with all accrued tokens at cancellation
// Effects: update all stream state before any external call
if accrued_amount > 0 {
token_client.transfer(&contract_address, &stream.recipient, &accrued_amount);
stream.withdrawn_amount = stream.withdrawn_amount.saturating_add(accrued_amount);
}

// Calculate and refund remaining balance to sender
let refunded_amount = stream
.deposited_amount
.saturating_sub(stream.withdrawn_amount);

if refunded_amount > 0 {
token_client.transfer(&contract_address, &sender, &refunded_amount);
}

// Mark stream as inactive
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();
let amount_withdrawn = stream.withdrawn_amount;

// Persist state before any external calls (CEI)
save_stream(&env, stream_id, &stream);

// Interactions: token transfers after state is committed to storage
let token_client = token::Client::new(&env, &stream.token_address);
let contract_address = env.current_contract_address();

if accrued_amount > 0 {
token_client.transfer(&contract_address, &recipient, &accrued_amount);
}

if refunded_amount > 0 {
token_client.transfer(&contract_address, &sender, &refunded_amount);
}

// Emit cancellation event
env.events().publish(
(Symbol::new(&env, "stream_cancelled"), stream_id),
Expand All @@ -540,9 +551,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();

Expand Down Expand Up @@ -579,9 +590,9 @@ 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();

Expand Down Expand Up @@ -622,7 +633,7 @@ impl StreamContract {
Ok(new_end_time)
}

// ─── Read-only Queries ────────────────────────────────────────────────────
// ΓöÇΓöÇΓöÇ Read-only Queries ΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇΓöÇ

/// 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> {
Expand Down Expand Up @@ -652,7 +663,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.
Expand Down
Loading