diff --git a/.github/workflows/pr-test-gate.yml b/.github/workflows/pr-test-gate.yml index efe5c66d..b2471627 100644 --- a/.github/workflows/pr-test-gate.yml +++ b/.github/workflows/pr-test-gate.yml @@ -53,6 +53,9 @@ jobs: - name: Install dependencies run: npm ci --include=optional + - name: Install Rollup Native Binding + run: npm install @rollup/rollup-linux-x64-gnu --no-save + - name: Prepare database schema run: | npx prisma generate --schema=prisma/schema.prisma diff --git a/contracts/stream_contract/src/lib.rs b/contracts/stream_contract/src/lib.rs index 6f4875f9..3d0b0acc 100644 --- a/contracts/stream_contract/src/lib.rs +++ b/contracts/stream_contract/src/lib.rs @@ -253,6 +253,12 @@ impl StreamContract { /// /// Excludes any time the stream was paused. If the stream is currently /// paused, accrual stops at `paused_at`. + /// + /// # Overflow Protection + /// - Uses `checked_mul` for rate_per_second * elapsed_seconds multiplication + /// - Caps at stream.deposited_amount if overflow would occur + /// - Uses `checked_sub` for deposited - already_withdrawn calculation + /// - Overflow boundary: i128::MAX (~1.7e19) for both rate and duration fn calculate_claimable(stream: &Stream, now: u64) -> i128 { let effective_now = if stream.paused { stream.paused_at.unwrap_or(stream.last_update_time) @@ -261,13 +267,21 @@ impl StreamContract { }; let elapsed = effective_now.saturating_sub(stream.last_update_time); - let streamed = (elapsed as i128) - .checked_mul(stream.rate_per_second) - .unwrap_or(i128::MAX); + // Use checked_mul to prevent overflow when multiplying rate * elapsed + // If overflow would occur, cap at deposited_amount (full deposit) + let streamed = match (elapsed as i128).checked_mul(stream.rate_per_second) { + Some(result) => result, + None => return stream.deposited_amount, // Overflow: cap at full deposit + }; - let remaining = stream + // Use checked_sub for deposited - withdrawn calculation + let remaining = match stream .deposited_amount - .saturating_sub(stream.withdrawn_amount); + .checked_sub(stream.withdrawn_amount) + { + Some(result) => result, + None => 0, // Underflow: already withdrawn more than deposited + }; streamed.min(remaining) } diff --git a/contracts/stream_contract/src/test.rs b/contracts/stream_contract/src/test.rs index 6d22901c..d511f807 100644 --- a/contracts/stream_contract/src/test.rs +++ b/contracts/stream_contract/src/test.rs @@ -948,6 +948,43 @@ fn test_cancel_stream_after_partial_withdrawal() { assert_eq!(contract_balance_after, 0); } +#[test] +fn test_claimable_max_i128_rate_overflow() { + 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, i128::MAX); + + let client = create_contract(&env); + + // Create stream with near-max i128 rate + let max_rate = i128::MAX / 2; + let stream_id = client.create_stream(&sender, &recipient, &token, &1_000, &1); + + // Manually set rate to near-max i128 to test overflow protection + let mut stream = client.get_stream(&stream_id).unwrap(); + stream.rate_per_second = max_rate; + env.as_contract(&client.address, || { + env.storage().persistent().set(&types::DataKey::Stream(stream_id), &stream); + }); + + // Advance time by a large amount that would cause overflow + env.ledger().with_mut(|l| { + l.timestamp += 1_000_000_000; + }); + + // get_claimable_amount should cap at deposited_amount, not overflow + let claimable = client.get_claimable_amount(&stream_id).unwrap(); + assert_eq!(claimable, 1_000); // Should cap at deposited amount + + // Withdraw should work correctly without overflow + let withdrawn = client.withdraw(&recipient, &stream_id); + assert_eq!(withdrawn, 1_000); +} + // ─── #232 create_stream edge cases ─────────────────────────────────────────── #[test] diff --git a/frontend/app/streams/[id]/page.tsx b/frontend/app/streams/[id]/page.tsx index 03f933a9..32f4c59a 100644 --- a/frontend/app/streams/[id]/page.tsx +++ b/frontend/app/streams/[id]/page.tsx @@ -1,62 +1,174 @@ +"use client"; -import { notFound } from "next/navigation"; +import { useEffect, useState } from "react"; +import { useParams } from "next/navigation"; import LiveCounter from "@/components/Livecounter"; import ProgressBar from "@/components/Progressbar"; +import { Button } from "@/components/ui/Button"; +import toast from "react-hot-toast"; +import { + withdrawFromStream, + cancelStream, + topUpStream, + toSorobanErrorMessage, +} from "@/lib/soroban"; +import type { WalletSession } from "@/lib/wallet"; - -interface Transaction { +interface StreamDetail { id: string; - date: string; - amount: number; - type: "withdrawal"; -} - -interface Stream { - id: string; - name: string; + sender: string; recipient: string; - streamedAmount: number; - totalAmount: number; - transactions: Transaction[]; + tokenAddress: string; + depositedAmount: string; + withdrawnAmount: string; + ratePerSecond: string; + startTime: number; + lastUpdateTime: number; + isActive: boolean; + status: string; } -const MOCK_STREAMS: Record = { - "1": { - id: "1", - name: "Developer Grant — Q1", - recipient: "0xAbC…1234", - streamedAmount: 3200, - totalAmount: 5000, - transactions: [ - { id: "tx1", date: "2026-02-20", amount: 1000, type: "withdrawal" }, - { id: "tx2", date: "2026-02-18", amount: 1200, type: "withdrawal" }, - { id: "tx3", date: "2026-02-15", amount: 1000, type: "withdrawal" }, - ], - }, - "2": { - id: "2", - name: "Marketing Budget Stream", - recipient: "0xDeF…5678", - streamedAmount: 800, - totalAmount: 2000, - transactions: [ - { id: "tx1", date: "2026-02-21", amount: 500, type: "withdrawal" }, - { id: "tx2", date: "2026-02-19", amount: 300, type: "withdrawal" }, - ], - }, -}; - -interface PageProps { - params: Promise<{ id: string }>; -} +export default function StreamDetailsPage() { + const params = useParams(); + const streamId = params.id as string; + + const [stream, setStream] = useState(null); + const [loading, setLoading] = useState(true); + const [error, setError] = useState(null); + const [withdrawing, setWithdrawing] = useState(false); + const [cancelling, setCancelling] = useState(false); + const [topUpAmount, setTopUpAmount] = useState(""); + const [showTopUp, setShowTopUp] = useState(false); + const [session, setSession] = useState(null); + + // Mock session - in production this would come from a wallet context + useEffect(() => { + // This would normally come from a wallet provider context + // For now, we'll use a mock session + setSession({ + walletId: "freighter", + publicKey: "GD...", + network: "TESTNET", + walletName: "Freighter", + connectedAt: new Date().toISOString(), + mocked: true, + }); + }, []); + + useEffect(() => { + async function fetchStream() { + try { + const baseUrl = process.env.NEXT_PUBLIC_API_URL || "http://localhost:3001"; + const response = await fetch(`${baseUrl}/v1/streams/${streamId}`); + if (!response.ok) { + throw new Error("Stream not found"); + } + const data = await response.json(); + setStream(data); + } catch (err) { + setError(err instanceof Error ? err.message : "Failed to fetch stream"); + } finally { + setLoading(false); + } + } + + if (streamId) { + fetchStream(); + } + }, [streamId]); + + const handleWithdraw = async () => { + if (!session) { + toast.error("Please connect your wallet first"); + return; + } -export default async function StreamDetailsPage({ params }: PageProps) { - const { id } = await params; - const stream = MOCK_STREAMS[id]; + setWithdrawing(true); + try { + await withdrawFromStream(session, { streamId: BigInt(streamId) }); + toast.success("Withdrawal successful!"); + // Refresh stream data + window.location.reload(); + } catch (err) { + toast.error(toSorobanErrorMessage(err)); + } finally { + setWithdrawing(false); + } + }; + + const handleCancel = async () => { + if (!session) { + toast.error("Please connect your wallet first"); + return; + } + + if (!confirm("Are you sure you want to cancel this stream?")) { + return; + } + + setCancelling(true); + try { + await cancelStream(session, { streamId: BigInt(streamId) }); + toast.success("Stream cancelled successfully!"); + // Refresh stream data + window.location.reload(); + } catch (err) { + toast.error(toSorobanErrorMessage(err)); + } finally { + setCancelling(false); + } + }; + + const handleTopUp = async () => { + if (!session) { + toast.error("Please connect your wallet first"); + return; + } + + if (!topUpAmount || parseFloat(topUpAmount) <= 0) { + toast.error("Please enter a valid amount"); + return; + } + + try { + await topUpStream(session, { + streamId: BigInt(streamId), + amount: BigInt(parseFloat(topUpAmount) * 1e7), // Convert to stroops + }); + toast.success("Stream topped up successfully!"); + setShowTopUp(false); + setTopUpAmount(""); + // Refresh stream data + window.location.reload(); + } catch (err) { + toast.error(toSorobanErrorMessage(err)); + } + }; + + if (loading) { + return ( +
+
+

Loading stream details...

+
+
+ ); + } - if (!stream) notFound(); + if (error || !stream) { + return ( +
+
+

{error || "Stream not found"}

+
+
+ ); + } - const percentage = Math.round((stream.streamedAmount / stream.totalAmount) * 100); + const deposited = parseFloat(stream.depositedAmount) / 1e7; + const withdrawn = parseFloat(stream.withdrawnAmount) / 1e7; + const claimable = deposited - withdrawn; + const percentage = Math.round((withdrawn / deposited) * 100); return (
@@ -64,7 +176,7 @@ export default async function StreamDetailsPage({ params }: PageProps) { {/* Header */}
-

Stream #{stream.id}

+

Stream #{streamId}

Stream Details

@@ -72,70 +184,134 @@ export default async function StreamDetailsPage({ params }: PageProps) { {/* Identity card */}
-

{stream.name}

-

- Recipient:{" "} - - {stream.recipient} - -

+
+
+

+ Status: {stream.status} +

+

+ Sender:{" "} + + {stream.sender.slice(0, 8)}...{stream.sender.slice(-4)} + +

+

+ Recipient:{" "} + + {stream.recipient.slice(0, 8)}...{stream.recipient.slice(-4)} + +

+

+ Token: {stream.tokenAddress.slice(0, 8)}... +

+
+
+

+ Rate: {(parseFloat(stream.ratePerSecond) / 1e7).toFixed(7)} / sec +

+

+ Started: {new Date(stream.startTime * 1000).toLocaleDateString()} +

+
+
- {/* 1️⃣ Progress bar */} + {/* Progress bar */}
-

Streamed Progress

+

Stream Progress

- {/* 3️⃣ Live counter */} + {/* Live counter */}
-

Live Balance

+

Claimable Balance

- +
- {/* 2️⃣ Transaction history */} + {/* Actions */}
-

Transaction History

- {stream.transactions.length} withdrawals +

Actions

+
+
+ + +
- {stream.transactions.length === 0 ? ( -
-

No transactions yet.

+ {showTopUp && ( +
+ setTopUpAmount(e.target.value)} + style={{ + padding: "0.5rem", + borderRadius: "0.25rem", + border: "1px solid var(--glass-border)", + background: "rgba(255,255,255,0.05)", + color: "inherit", + }} + /> +
- ) : ( -
    - {stream.transactions.map((tx) => ( -
  • -
    - Withdrawal -

    {tx.date}

    -
    - - -{tx.amount.toLocaleString()} tokens - -
    • - ))} -
)}
+ {/* Transaction history */} +
+
+

Transaction History

+ Recent activity +
+
+

Transaction history will be populated from backend events.

+
+
+
); diff --git a/frontend/components/dashboard/dashboard-view.tsx b/frontend/components/dashboard/dashboard-view.tsx index f1440bd6..abbc6c1e 100644 --- a/frontend/components/dashboard/dashboard-view.tsx +++ b/frontend/components/dashboard/dashboard-view.tsx @@ -37,6 +37,7 @@ import { toSorobanErrorMessage, } from "@/lib/soroban"; import IncomingStreams from "../IncomingStreams"; +import { useStreamEvents } from "@/hooks/useStreamEvents"; import { StreamCreationWizard, type StreamFormData, @@ -327,6 +328,30 @@ export function DashboardView({ session, onDisconnect }: DashboardViewProps) { const [showWizard, setShowWizard] = React.useState(false); const [modal, setModal] = React.useState(null); + // SSE integration for real-time stream updates + const { events: streamEvents } = useStreamEvents({ + userPublicKeys: [session.publicKey], + autoReconnect: true, + }); + + // Refresh dashboard when SSE events arrive + React.useEffect(() => { + if (streamEvents.length > 0) { + const latestEvent = streamEvents[0]; + console.log('SSE event received:', latestEvent); + // Refresh dashboard data on relevant events + if (latestEvent.type === 'created' || latestEvent.type === 'topped_up' || + latestEvent.type === 'withdrawn' || latestEvent.type === 'cancelled' || + latestEvent.type === 'completed') { + fetchDashboardData(session.publicKey) + .then(setSnapshot) + .catch(err => { + setSnapshotError(err instanceof Error ? err.message : 'Failed to refresh dashboard'); + }); + } + } + }, [streamEvents, session.publicKey]); + // --- Templates State (from upstream) --- const [streamForm, setStreamForm] = React.useState( EMPTY_STREAM_FORM, diff --git a/frontend/hooks/useStreamEvents.ts b/frontend/hooks/useStreamEvents.ts new file mode 100644 index 00000000..4b3a9967 --- /dev/null +++ b/frontend/hooks/useStreamEvents.ts @@ -0,0 +1,149 @@ +import { useEffect, useState, useCallback, useRef } from 'react'; + +interface StreamEvent { + type: 'created' | 'topped_up' | 'withdrawn' | 'cancelled' | 'completed'; + data: unknown; + timestamp: number; +} + +interface UseStreamEventsOptions { + streamIds?: string[]; + userPublicKeys?: string[]; + subscribeToAll?: boolean; + autoReconnect?: boolean; + maxRetryDelay?: number; +} + +interface UseStreamEventsReturn { + events: StreamEvent[]; + connected: boolean; + error: Error | null; + reconnecting: boolean; + clearEvents: () => void; +} + +export function useStreamEvents( + options: UseStreamEventsOptions = {} +): UseStreamEventsReturn { + const { + streamIds = [], + userPublicKeys = [], + subscribeToAll = false, + autoReconnect = true, + maxRetryDelay = 30000, + } = options; + + const [events, setEvents] = useState([]); + const [connected, setConnected] = useState(false); + const [error, setError] = useState(null); + const [reconnecting, setReconnecting] = useState(false); + + const eventSourceRef = useRef(null); + const retryDelayRef = useRef(1000); + const reconnectTimeoutRef = useRef | null>(null); + const connectRef = useRef<() => void>(() => undefined); + + const buildUrl = useCallback(() => { + const params = new URLSearchParams(); + + if (subscribeToAll) { + params.append('all', 'true'); + } else { + streamIds.forEach(id => params.append('streams', id)); + userPublicKeys.forEach(key => params.append('users', key)); + } + + const baseUrl = process.env.NEXT_PUBLIC_API_URL || 'http://localhost:3001'; + return `${baseUrl}/events/subscribe?${params}`; + }, [streamIds, userPublicKeys, subscribeToAll]); + + const clearEvents = useCallback(() => { + setEvents([]); + }, []); + + const connect = useCallback(() => { + const url = buildUrl(); + const eventSource = new EventSource(url); + eventSourceRef.current = eventSource; + + eventSource.onopen = () => { + setConnected(true); + setReconnecting(false); + setError(null); + retryDelayRef.current = 1000; // Reset retry delay + }; + + eventSource.onmessage = (e) => { + try { + const data = JSON.parse(e.data); + if (data.type === 'connected') { + console.log('SSE connected:', data.clientId); + } + } catch (err) { + console.error('Failed to parse SSE message:', err); + } + }; + + const handleEvent = (type: StreamEvent['type']) => (e: MessageEvent) => { + try { + const data = JSON.parse(e.data); + setEvents((prev: StreamEvent[]) => [ + { type, data, timestamp: Date.now() }, + ...prev.slice(0, 99), // Keep last 100 events + ]); + } catch (err) { + console.error(`Failed to parse ${type} event:`, err); + } + }; + + eventSource.addEventListener('stream.created', handleEvent('created')); + eventSource.addEventListener('stream.topped_up', handleEvent('topped_up')); + eventSource.addEventListener('stream.withdrawn', handleEvent('withdrawn')); + eventSource.addEventListener('stream.cancelled', handleEvent('cancelled')); + eventSource.addEventListener('stream.completed', handleEvent('completed')); + + eventSource.onerror = () => { + setConnected(false); + setError(new Error('SSE connection failed')); + eventSource.close(); + + if (autoReconnect) { + setReconnecting(true); + reconnectTimeoutRef.current = setTimeout(() => { + console.log(`Reconnecting in ${retryDelayRef.current}ms...`); + connectRef.current(); + retryDelayRef.current = Math.min( + retryDelayRef.current * 2, + maxRetryDelay + ); + }, retryDelayRef.current); + } + }; + }, [buildUrl, autoReconnect, maxRetryDelay]); + + useEffect(() => { + connectRef.current = connect; + }, [connect]); + + useEffect(() => { + connect(); + + return () => { + if (eventSourceRef.current) { + eventSourceRef.current.close(); + eventSourceRef.current = null; + } + if (reconnectTimeoutRef.current) { + clearTimeout(reconnectTimeoutRef.current); + } + }; + }, [connect]); + + return { + events, + connected, + error, + reconnecting, + clearEvents, + }; +}