diff --git a/app/dashboard/page.tsx b/app/dashboard/page.tsx index 5f18013..9fddb55 100644 --- a/app/dashboard/page.tsx +++ b/app/dashboard/page.tsx @@ -31,12 +31,17 @@ function deriveStatus(info: StreamInfo, now: number): StreamStatus { return "active"; } +interface LoadRowsResult { + rows: StreamRow[]; + failedCount: number; +} + async function loadRows( publicKey: string, role: "sender" | "recipient", now: number, signal: AbortSignal, -): Promise { +): Promise { let ids: bigint[]; try { ids = @@ -44,41 +49,68 @@ async function loadRows( ? await streamsBySender(publicKey, publicKey, 0, 50, { signal }) : await streamsByRecipient(publicKey, publicKey, 0, 50, { signal }); } catch { - return []; + return { rows: [], failedCount: 0 }; } - if (!ids || !Array.isArray(ids)) return []; + if (!ids || !Array.isArray(ids)) return { rows: [], failedCount: 0 }; + + const uniqueIds = [...new Set(ids.filter((id): id is bigint => typeof id === "bigint"))]; + + // Phase 1: resolve all stream addresses in parallel + const addrResults = await Promise.allSettled( + uniqueIds.map((id) => getStreamAddress(publicKey, id, { signal })), + ); + + const addrPairs: { id: bigint; rowId: string; addr: string }[] = []; + let failedCount = 0; + for (let i = 0; i < uniqueIds.length; i++) { + const r = addrResults[i]; + if (signal.aborted) return { rows: [], failedCount: 0 }; + if (r.status === "fulfilled" && r.value && typeof r.value === "string") { + addrPairs.push({ id: uniqueIds[i]!, rowId: uniqueIds[i]!.toString(), addr: r.value }); + } else { + failedCount++; + } + } + // Phase 2: fetch info+withdrawable in bounded-parallel batches + const BATCH_SIZE = 5; const rows: StreamRow[] = []; - const seen = new Set(); - for (const id of ids) { - if (signal.aborted) return []; - if (typeof id !== "bigint") continue; - const rowId = id.toString(); - if (seen.has(rowId)) continue; - try { - const addr = await getStreamAddress(publicKey, id, { signal }); - if (!addr || typeof addr !== "string") continue; - const [info, withdrawable] = await Promise.all([ - getStreamInfo(publicKey, addr, { signal }), - getWithdrawable(publicKey, addr, { signal }), - ]); - if (!info || typeof info !== "object") continue; - if (typeof info.ratePerSecond !== "bigint") continue; - if (signal.aborted) return []; - rows.push({ - id: rowId, - address: addr, - info, - withdrawable, - status: deriveStatus(info, now), - }); - seen.add(rowId); - } catch { - /* skip invalid streams */ + + for (let start = 0; start < addrPairs.length; start += BATCH_SIZE) { + if (signal.aborted) return { rows: [], failedCount: 0 }; + const batch = addrPairs.slice(start, start + BATCH_SIZE); + const results = await Promise.allSettled( + batch.map(({ addr }) => + Promise.all([ + getStreamInfo(publicKey, addr, { signal }), + getWithdrawable(publicKey, addr, { signal }), + ]), + ), + ); + for (let j = 0; j < results.length; j++) { + const r = results[j]; + const pair = batch[j]!; + if ( + r.status === "fulfilled" && + r.value[0] && + typeof r.value[0] === "object" && + typeof r.value[0].ratePerSecond === "bigint" + ) { + rows.push({ + id: pair.rowId, + address: pair.addr, + info: r.value[0], + withdrawable: r.value[1], + status: deriveStatus(r.value[0], now), + }); + } else { + failedCount++; + } } } - return rows; + + return { rows, failedCount }; } // ── Page ────────────────────────────────────────────────────────────────────── @@ -91,6 +123,7 @@ export default function DashboardPage() { const [sending, setSending] = useState([]); const [loading, setLoading] = useState(false); const [error, setError] = useState(null); + const [partialError, setPartialError] = useState(null); // #307 — loadSeqRef is the ordering guard: each fetch captures its own // seq and only commits state if it's still the most recent request by the // time it resolves, the same pattern app/stream/[id]/page.tsx uses. @@ -100,6 +133,7 @@ export default function DashboardPage() { // one, instead of each site spinning up its own untracked controller. const loadSeqRef = useRef(0); const activeControllerRef = useRef(null); + const lastFetchAtRef = useRef(0); const fetchStreams = useCallback(async (signal: AbortSignal) => { if (!publicKey) return; @@ -108,14 +142,22 @@ export default function DashboardPage() { const now = Math.floor(Date.now() / 1000); setLoading(true); setError(null); + setPartialError(null); try { const [recv, sent] = await Promise.all([ loadRows(publicKey, "recipient", now, signal), loadRows(publicKey, "sender", now, signal), ]); if (!signal.aborted && isCurrent()) { - setReceiving(recv); - setSending(sent); + setReceiving(recv.rows); + setSending(sent.rows); + const totalFailed = recv.failedCount + sent.failedCount; + if (totalFailed > 0) { + setPartialError( + `${totalFailed} stream${totalFailed === 1 ? "" : "s"} could not be loaded — some data may be missing.`, + ); + } + lastFetchAtRef.current = Date.now(); } } catch (e) { if (!signal.aborted && isCurrent()) { @@ -139,12 +181,15 @@ export default function DashboardPage() { setReceiving([]); setSending([]); setError(null); + setPartialError(null); return; } refetch(); const handleVisibilityChange = () => { if (document.visibilityState === 'visible') { + // Skip refetch if the last fetch was less than 5s ago + if (Date.now() - lastFetchAtRef.current < 5_000) return; refetch(); } }; @@ -237,6 +282,19 @@ export default function DashboardPage() { )} + {partialError && !error && ( +
+
+ )} + {!connected ? (
Connect your wallet to see your streams. diff --git a/contexts/WalletContext.tsx b/contexts/WalletContext.tsx index 92533ae..8a0def1 100644 --- a/contexts/WalletContext.tsx +++ b/contexts/WalletContext.tsx @@ -33,6 +33,7 @@ import { } from '@stellar/freighter-api'; import { getNetworkPassphrase } from '@/lib/env'; import { queryClient } from '@/lib/queryClient'; +import { resetTokenAllowanceGateway } from '@/lib/token-allowance-gateway'; import { useTransactionStore } from '@/lib/store'; import { truncateAddress } from '@/lib/format'; import { useRouter } from 'next/navigation'; @@ -440,6 +441,7 @@ export function WalletProvider({ // Clear all cached stream data so a subsequent wallet connection // cannot see the previous wallet's streams (fixes #81 & #146). queryClient.clear(); + resetTokenAllowanceGateway(); clearTransactions(); router.push('/'); }, [clearTransactions, router]); @@ -470,6 +472,7 @@ export function WalletProvider({ setPublicKey(address); saveWalletSession({ key: address, name: 'Freighter' }); queryClient.clear(); + resetTokenAllowanceGateway(); clearTransactions(); toast(`Switched to ${truncateAddress(address)}`, { icon: '🔄' }); }); diff --git a/lib/token-allowance-gateway.ts b/lib/token-allowance-gateway.ts index 2d67fa1..b85358b 100644 --- a/lib/token-allowance-gateway.ts +++ b/lib/token-allowance-gateway.ts @@ -195,7 +195,7 @@ export interface RevokeAllowanceArgs { */ export class TokenAllowanceGateway { private _records = new Map(); - private _concurrencySemaphore: { available: number; queue: Array<() => void> } = { + private _concurrencySemaphore: { available: number; queue: Array<{ resolve: () => void; reject: (err: Error) => void }> } = { available: 5, queue: [], }; @@ -210,12 +210,12 @@ export class TokenAllowanceGateway { // ── Record management ───────────────────────────────────────────────────── - private _key(token: string, spender: string): string { - return `${token}::${spender}`; + private _key(owner: string, token: string, spender: string): string { + return `${owner}::${token}::${spender}`; } - private _getOrCreate(token: string, spender: string): InternalRecord { - const key = this._key(token, spender); + private _getOrCreate(owner: string, token: string, spender: string): InternalRecord { + const key = this._key(owner, token, spender); let record = this._records.get(key); if (!record) { record = { @@ -229,10 +229,10 @@ export class TokenAllowanceGateway { } /** - * Read-only snapshot of the current allowance record for a token+spender pair. + * Read-only snapshot of the current allowance record for an owner+token+spender triple. */ - getAllowance(token: string, spender: string): AllowanceRecord { - const key = this._key(token, spender); + getAllowance(owner: string, token: string, spender: string): AllowanceRecord { + const key = this._key(owner, token, spender); const record = this._records.get(key); if (!record) { return { allowance: 0n, state: 'idle' }; @@ -257,12 +257,15 @@ export class TokenAllowanceGateway { let settled = false; let cleanup: (() => void) | undefined; - const entry = () => { - if (!settled) { - settled = true; - cleanup?.(); - resolve(() => this._releaseConcurrency()); - } + const entry = { + resolve: () => { + if (!settled) { + settled = true; + cleanup?.(); + resolve(() => this._releaseConcurrency()); + } + }, + reject, }; this._concurrencySemaphore.queue.push(entry); @@ -294,7 +297,7 @@ export class TokenAllowanceGateway { private _releaseConcurrency() { const next = this._concurrencySemaphore.queue.shift(); if (next) { - next(); + next.resolve(); } else { this._concurrencySemaphore.available++; } @@ -312,7 +315,7 @@ export class TokenAllowanceGateway { */ async approve(args: ApproveAllowanceArgs): Promise> { const { token, spender, amount, source, signTx, signal } = args; - const record = this._getOrCreate(token, spender); + const record = this._getOrCreate(source, token, spender); const idempotencyKey = makeOperationKey(source, token, 'approve', spender, amount.toString()); // Reject if a previous operation is in-flight for this pair (unless same idempotency key) @@ -499,7 +502,7 @@ export class TokenAllowanceGateway { const allowance = scValToI128(result); // Update local cache - const record = this._getOrCreate(token, spender); + const record = this._getOrCreate(owner, token, spender); record.allowance = allowance; if (record.state === 'idle' || record.state === 'confirmed') { record.state = 'confirmed'; @@ -545,11 +548,11 @@ export class TokenAllowanceGateway { } this._records.clear(); - // Drain concurrency queue + // Drain concurrency queue — reject waiters so they don't proceed + // against a disconnected wallet while (this._concurrencySemaphore.queue.length > 0) { const entry = this._concurrencySemaphore.queue.shift(); - // Entries waiting for a slot get rejected — they'll retry on next call - entry?.(); + entry?.reject(new OperationAbortedError('Gateway reset — wallet disconnected')); } this._concurrencySemaphore.available = this._maxConcurrency; }