diff --git a/frontend/src/__tests__/optimistic-ui.test.tsx b/frontend/src/__tests__/optimistic-ui.test.tsx new file mode 100644 index 00000000..5893fc02 --- /dev/null +++ b/frontend/src/__tests__/optimistic-ui.test.tsx @@ -0,0 +1,159 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest'; +import { render, screen, waitFor, fireEvent, act } from '@testing-library/react'; +import type { CachedStream } from '@/lib/storage/stream-cache'; + +const mocks = vi.hoisted(() => { + const stream: CachedStream = { + id: 'stream-1', + sender: '0xSender', + recipient: '0xRecipient', + tokenAddress: '0xUSDC', + tokenSymbol: 'USDC', + tokenDecimals: 6, + depositedAmount: '1000000000', + withdrawnAmount: '0', + claimableAmount: '500000000', + status: 'active', + updatedAt: 1_700_000_000_000, + }; + return { + stream, + getCachedStreams: vi.fn(async () => [stream]), + getOptimisticMutations: vi.fn(async () => []), + putCachedStream: vi.fn(async () => undefined), + putCachedStreams: vi.fn(async () => undefined), + putOptimisticMutation: vi.fn(async () => undefined), + deleteOptimisticMutation: vi.fn(async () => undefined), + }; +}); + +vi.mock('@/lib/storage/stream-cache', () => ({ + getCachedStreams: mocks.getCachedStreams, + getOptimisticMutations: mocks.getOptimisticMutations, + putCachedStream: mocks.putCachedStream, + putCachedStreams: mocks.putCachedStreams, + putOptimisticMutation: mocks.putOptimisticMutation, + deleteOptimisticMutation: mocks.deleteOptimisticMutation, +})); + +import { StreamProvider } from '@/context/StreamContext'; +import { StreamCard } from '@/components/dashboard/StreamCard'; + +function renderCard(commit?: (ctx: unknown) => Promise) { + return render( + + + , + ); +} + +describe('optimistic ui + IndexedDB cache', () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + it('hydrates from IndexedDB immediately without a blocking spinner', async () => { + renderCard(); + const card = await screen.findByTestId('stream-card-stream-1'); + expect(card).toBeInTheDocument(); + expect(screen.queryByTestId('stream-skeleton')).not.toBeInTheDocument(); + expect(mocks.getCachedStreams).toHaveBeenCalledTimes(1); + expect(mocks.getOptimisticMutations).toHaveBeenCalledTimes(1); + }); + + it('optimistically updates withdrawn + claimable on withdraw', async () => { + const commit = vi.fn(async () => ({})); + renderCard(commit); + + await screen.findByTestId('stream-card-stream-1'); + expect(screen.getByTestId('stream-withdrawn').textContent).toMatch(/^0$/); + expect(screen.getByTestId('stream-claimable').textContent).toMatch(/^500$/); + + await act(async () => { + fireEvent.click(screen.getByTestId('stream-withdraw-btn')); + }); + + await waitFor(() => { + expect(screen.getByTestId('stream-withdrawn').textContent).toMatch(/^500$/); + }); + expect(screen.getByTestId('stream-claimable').textContent).toMatch(/^0$/); + }); + + it('shows the pending ledger badge while the mutation is in flight', async () => { + let resolveCommit: (value: unknown) => void = () => undefined; + const commit = vi.fn( + () => + new Promise((resolve) => { + resolveCommit = resolve; + }), + ); + renderCard(commit); + + await screen.findByTestId('stream-card-stream-1'); + await act(async () => { + fireEvent.click(screen.getByTestId('stream-withdraw-btn')); + }); + + expect(await screen.findByTestId('pending-badge')).toHaveTextContent( + /Pending Ledger Confirmation/i, + ); + + await act(async () => { + resolveCommit({}); + }); + + await waitFor(() => { + expect(screen.queryByTestId('pending-badge')).not.toBeInTheDocument(); + }); + expect(screen.getByTestId('stream-withdrawn').textContent).toMatch(/^500$/); + }); + + it('rolls back on transaction failure and surfaces an error', async () => { + const commit = vi.fn(async () => { + throw new Error('reverted on-chain'); + }); + renderCard(commit); + + await screen.findByTestId('stream-card-stream-1'); + await act(async () => { + fireEvent.click(screen.getByTestId('stream-withdraw-btn')); + }); + + expect(await screen.findByTestId('stream-error')).toHaveTextContent(/reverted on-chain/i); + + await waitFor(() => { + expect(screen.getByTestId('stream-withdrawn').textContent).toMatch(/^0$/); + }); + expect(screen.getByTestId('stream-claimable').textContent).toMatch(/^500$/); + expect(screen.queryByTestId('pending-badge')).not.toBeInTheDocument(); + }); + + it('cancels the stream optimistically', async () => { + let resolveCommit: (value: unknown) => void = () => undefined; + const commit = vi.fn( + () => + new Promise((resolve) => { + resolveCommit = resolve; + }), + ); + renderCard(commit); + + await screen.findByTestId('stream-card-stream-1'); + await act(async () => { + fireEvent.click(screen.getByTestId('stream-cancel-btn')); + }); + + // Badge visible while pending. + expect(await screen.findByTestId('pending-badge')).toBeInTheDocument(); + + // Resolve — badge clears, card remains. + await act(async () => { + resolveCommit({}); + }); + + await waitFor(() => { + expect(screen.queryByTestId('pending-badge')).not.toBeInTheDocument(); + }); + expect(screen.getByTestId('stream-card-stream-1')).toBeInTheDocument(); + }); +}); diff --git a/frontend/src/components/dashboard/StreamCard.tsx b/frontend/src/components/dashboard/StreamCard.tsx new file mode 100644 index 00000000..3cc67a6e --- /dev/null +++ b/frontend/src/components/dashboard/StreamCard.tsx @@ -0,0 +1,141 @@ +'use client'; + +import { + useOptimisticStream, + type CommitContext, + type CommitResult, +} from '@/hooks/useOptimisticStream'; + +export interface StreamCardProps { + streamId: string; + commit?: (ctx: CommitContext) => Promise; + onError?: (message: string) => void; +} + +function formatAmount(raw: string, decimals: number): string { + try { + const value = BigInt(raw); + const base = 10n ** BigInt(decimals); + const whole = value / base; + const fraction = value % base; + if (fraction === 0n) return whole.toString(); + const frac = fraction.toString().padStart(decimals, '0').replace(/0+$/, ''); + return frac ? `${whole}.${frac}` : whole.toString(); + } catch { + return raw; + } +} + +export function StreamCard({ streamId, commit, onError }: StreamCardProps) { + const { stream, isPending, error, withdraw, topUp, cancel, pause } = useOptimisticStream( + streamId, + { commit, onError }, + ); + + if (!stream) { + return ( +
+

Loading stream…

+
+ ); + } + + return ( +
+
+
+

+ {stream.tokenSymbol} stream +

+

{stream.id}

+
+ + {isPending ? ( + + + Pending Ledger Confirmation + + ) : null} +
+ +
+
+
Deposited
+
+ {formatAmount(stream.depositedAmount, stream.tokenDecimals)} +
+
+
+
Withdrawn
+
+ {formatAmount(stream.withdrawnAmount, stream.tokenDecimals)} +
+
+
+
Claimable
+
+ {formatAmount(stream.claimableAmount, stream.tokenDecimals)} +
+
+
+ + {error ? ( +

+ {error} +

+ ) : null} + +
+ + + + +
+
+ ); +} + +export default StreamCard; diff --git a/frontend/src/context/StreamContext.tsx b/frontend/src/context/StreamContext.tsx new file mode 100644 index 00000000..6849c69b --- /dev/null +++ b/frontend/src/context/StreamContext.tsx @@ -0,0 +1,239 @@ +'use client'; + +import { + createContext, + useCallback, + useContext, + useEffect, + useMemo, + useReducer, + type ReactNode, +} from 'react'; +import { + type CachedStream, + type OptimisticMutation, + getCachedStreams, + getOptimisticMutations, + putCachedStreams, + putOptimisticMutation, + deleteOptimisticMutation, +} from '@/lib/storage/stream-cache'; + +export interface StreamContextValue { + /** Effective streams: cached canonical state + pending optimistic patches. */ + streams: CachedStream[]; + mutations: OptimisticMutation[]; + /** True once the IndexedDB hydration effect has run. */ + isHydrated: boolean; + /** True while a background network refresh is in flight. */ + isLoading: boolean; + error: Error | null; + getStream: (id: string) => CachedStream | undefined; + upsertStreams: (streams: CachedStream[]) => void; + applyOptimistic: (mutation: OptimisticMutation) => Promise; + resolveMutation: (mutationId: string) => Promise; + rollbackMutation: (mutationId: string) => Promise; + refresh: () => Promise; +} + +interface State { + streams: Record; + mutations: OptimisticMutation[]; + isHydrated: boolean; + isLoading: boolean; + error: Error | null; +} + +type Action = + | { type: 'hydrate'; streams: CachedStream[]; mutations: OptimisticMutation[] } + | { type: 'upsertStreams'; streams: CachedStream[] } + | { type: 'addMutation'; mutation: OptimisticMutation } + | { type: 'removeMutation'; id: string } + | { type: 'setLoading'; value: boolean } + | { type: 'setError'; error: Error | null }; + +const initialState: State = { + streams: {}, + mutations: [], + isHydrated: false, + isLoading: false, + error: null, +}; + +function reducer(state: State, action: Action): State { + switch (action.type) { + case 'hydrate': { + const streams: Record = {}; + for (const s of action.streams) streams[s.id] = s; + return { + ...state, + streams, + // Only keep pending mutations across reloads; confirmed/failed are stale. + mutations: action.mutations.filter((m) => m.status === 'pending'), + isHydrated: true, + isLoading: false, + }; + } + case 'upsertStreams': { + const next = { ...state.streams }; + for (const s of action.streams) next[s.id] = s; + return { ...state, streams: next, isLoading: false }; + } + case 'addMutation': + return { ...state, mutations: [...state.mutations, action.mutation] }; + case 'removeMutation': + return { ...state, mutations: state.mutations.filter((m) => m.id !== action.id) }; + case 'setLoading': + return { ...state, isLoading: action.value }; + case 'setError': + return { ...state, error: action.error }; + default: + return state; + } +} + +const StreamContext = createContext(null); + +export interface StreamProviderProps { + children: ReactNode; + /** + * Optional background fetcher that returns canonical streams from the API / + * indexer. Called once on mount and then on every `syncInterval` tick. + */ + fetcher?: () => Promise; + /** Polling interval for reconciliation in ms. 0 disables periodic sync. */ + syncInterval?: number; +} + +export function StreamProvider({ children, fetcher, syncInterval = 0 }: StreamProviderProps) { + const [state, dispatch] = useReducer(reducer, initialState); + + // 1. Hydrate instantly from IndexedDB (target: < 15ms). + useEffect(() => { + let cancelled = false; + void (async () => { + let cached: CachedStream[] = []; + let mutations: OptimisticMutation[] = []; + try { + [cached, mutations] = await Promise.all([ + getCachedStreams(), + getOptimisticMutations(), + ]); + } catch { + // Missing IndexedDB is not fatal — the app still works online. + } + if (cancelled) return; + dispatch({ type: 'hydrate', streams: cached, mutations }); + })(); + return () => { + cancelled = true; + }; + }, []); + + // 2. Background reconciliation against the network. + const refresh = useCallback(async () => { + if (!fetcher) return; + dispatch({ type: 'setLoading', value: true }); + try { + const fresh = await fetcher(); + dispatch({ type: 'upsertStreams', streams: fresh }); + dispatch({ type: 'setError', error: null }); + // Persist so the next cold start is instant. + void putCachedStreams(fresh).catch(() => undefined); + } catch (err) { + dispatch({ + type: 'setError', + error: err instanceof Error ? err : new Error('Failed to refresh streams'), + }); + } finally { + dispatch({ type: 'setLoading', value: false }); + } + }, [fetcher]); + + useEffect(() => { + if (!fetcher) return undefined; + void refresh(); + if (syncInterval > 0) { + const id = setInterval(() => void refresh(), syncInterval); + return () => clearInterval(id); + } + return undefined; + }, [fetcher, refresh, syncInterval]); + + const upsertStreams = useCallback((streams: CachedStream[]) => { + dispatch({ type: 'upsertStreams', streams }); + }, []); + + const applyOptimistic = useCallback( + async (mutation: OptimisticMutation) => { + dispatch({ type: 'addMutation', mutation }); + await putOptimisticMutation(mutation).catch(() => undefined); + }, + [], + ); + + const resolveMutation = useCallback(async (mutationId: string) => { + dispatch({ type: 'removeMutation', id: mutationId }); + await deleteOptimisticMutation(mutationId).catch(() => undefined); + }, []); + + const rollbackMutation = useCallback(async (mutationId: string) => { + dispatch({ type: 'removeMutation', id: mutationId }); + await deleteOptimisticMutation(mutationId).catch(() => undefined); + }, []); + + // 3. Effective streams = canonical map with pending patches layered on top. + const effectiveStreams = useMemo(() => { + const map = new Map(); + for (const s of Object.values(state.streams)) map.set(s.id, s); + for (const m of state.mutations) { + if (m.status !== 'pending') continue; + const base = map.get(m.streamId); + if (!base) continue; + map.set(m.streamId, { ...base, ...m.patch }); + } + return Array.from(map.values()); + }, [state.streams, state.mutations]); + + const getStream = useCallback( + (id: string) => effectiveStreams.find((s) => s.id === id), + [effectiveStreams], + ); + + const value = useMemo( + () => ({ + streams: effectiveStreams, + mutations: state.mutations, + isHydrated: state.isHydrated, + isLoading: state.isLoading, + error: state.error, + getStream, + upsertStreams, + applyOptimistic, + resolveMutation, + rollbackMutation, + refresh, + }), + [ + effectiveStreams, + state.mutations, + state.isHydrated, + state.isLoading, + state.error, + getStream, + upsertStreams, + applyOptimistic, + resolveMutation, + rollbackMutation, + refresh, + ], + ); + + return {children}; +} + +export function useStreamContext(): StreamContextValue { + const ctx = useContext(StreamContext); + if (!ctx) throw new Error('useStreamContext must be used inside a '); + return ctx; +} diff --git a/frontend/src/hooks/useOptimisticStream.ts b/frontend/src/hooks/useOptimisticStream.ts new file mode 100644 index 00000000..c6e7ec00 --- /dev/null +++ b/frontend/src/hooks/useOptimisticStream.ts @@ -0,0 +1,158 @@ +'use client'; + +import { useCallback, useMemo, useState } from 'react'; +import { useStreamContext } from '@/context/StreamContext'; +import type { CachedStream, MutationKind, OptimisticMutation } from '@/lib/storage/stream-cache'; + +export interface CommitContext { + streamId: string; + kind: MutationKind; + amount?: string; +} + +export interface CommitResult { + txHash?: string; + stream?: CachedStream; +} + +export interface UseOptimisticStreamOptions { + commit?: (ctx: CommitContext) => Promise; + onError?: (message: string) => void; +} + +export interface UseOptimisticStreamResult { + stream: CachedStream | undefined; + isPending: boolean; + pendingMutations: OptimisticMutation[]; + error: string | null; + withdraw: (amount: string) => Promise; + topUp: (amount: string) => Promise; + cancel: () => Promise; + pause: () => Promise; +} + +function addAmounts(a: string, b: string): string { + try { return (BigInt(a) + BigInt(b)).toString(); } catch { return a; } +} + +function subAmounts(a: string, b: string): string { + try { + const result = BigInt(a) - BigInt(b); + return result < 0n ? '0' : result.toString(); + } catch { return a; } +} + +function makeMutationId(): string { + return `mut_${Date.now().toString(36)}_${Math.random().toString(36).slice(2, 8)}`; +} + +export function useOptimisticStream( + streamId: string, + options: UseOptimisticStreamOptions = {}, +): UseOptimisticStreamResult { + const ctx = useStreamContext(); + const [error, setError] = useState(null); + + const stream = ctx.getStream(streamId); + const pendingMutations = useMemo( + () => ctx.mutations.filter((m) => m.streamId === streamId && m.status === 'pending'), + [ctx.mutations, streamId], + ); + const isPending = pendingMutations.length > 0; + + const run = useCallback( + async ( + kind: MutationKind, + patch: Partial, + before: Partial, + amount?: string, + ) => { + if (!stream) return; + setError(null); + + const mutation: OptimisticMutation = { + id: makeMutationId(), + streamId, + kind, + createdAt: Date.now(), + status: 'pending', + before, + patch, + }; + + await ctx.applyOptimistic(mutation); + + try { + const result = options.commit + ? await options.commit({ streamId, kind, amount }) + : ({} as CommitResult); + + if (result.stream) { + // Canonical record returned — reconcile silently. + ctx.upsertStreams([result.stream]); + } else { + // No canonical record yet — fold the optimistic patch into the + // cache so the UI keeps the new value until the indexer catches up. + ctx.upsertStreams([{ ...stream, ...patch }]); + } + await ctx.resolveMutation(mutation.id); + } catch (err) { + const message = err instanceof Error ? err.message : 'Transaction failed'; + await ctx.rollbackMutation(mutation.id); + setError(message); + options.onError?.(message); + } + }, + [ctx, options, stream, streamId], + ); + + const withdraw = useCallback( + async (amount: string) => { + if (!stream) return; + await run( + 'WITHDRAW', + { + withdrawnAmount: addAmounts(stream.withdrawnAmount, amount), + claimableAmount: subAmounts(stream.claimableAmount, amount), + updatedAt: Date.now(), + }, + { + withdrawnAmount: stream.withdrawnAmount, + claimableAmount: stream.claimableAmount, + }, + amount, + ); + }, + [run, stream], + ); + + const topUp = useCallback( + async (amount: string) => { + if (!stream) return; + await run( + 'TOP_UP', + { + depositedAmount: addAmounts(stream.depositedAmount, amount), + updatedAt: Date.now(), + }, + { depositedAmount: stream.depositedAmount }, + amount, + ); + }, + [run, stream], + ); + + const cancel = useCallback(async () => { + if (!stream) return; + await run('CANCEL', { status: 'cancelled', updatedAt: Date.now() }, { status: stream.status }); + }, [run, stream]); + + const pause = useCallback(async () => { + if (!stream) return; + await run('PAUSE', { status: 'paused', updatedAt: Date.now() }, { status: stream.status }); + }, [run, stream]); + + return { stream, isPending, pendingMutations, error, withdraw, topUp, cancel, pause }; +} + +export default useOptimisticStream; diff --git a/frontend/src/lib/storage/stream-cache.ts b/frontend/src/lib/storage/stream-cache.ts new file mode 100644 index 00000000..db40fcca --- /dev/null +++ b/frontend/src/lib/storage/stream-cache.ts @@ -0,0 +1,174 @@ +/** + * IndexedDB persistence layer for the optimistic UI engine. + * + * Two stores: + * - cached_streams: keyed by `id`, indexed by sender / recipient / tokenAddress / updatedAt + * - optimistic_mutations: keyed by `id`, one entry per pending user action + */ + +export type StreamStatus = 'active' | 'paused' | 'completed' | 'cancelled'; +export type MutationKind = 'CREATE' | 'WITHDRAW' | 'TOP_UP' | 'CANCEL' | 'PAUSE'; +export type MutationStatus = 'pending' | 'confirmed' | 'failed'; + +export interface CachedStream { + id: string; + sender: string; + recipient: string; + tokenAddress: string; + tokenSymbol: string; + tokenDecimals: number; + /** Base-unit amounts as decimal strings (bigint-safe). */ + depositedAmount: string; + withdrawnAmount: string; + claimableAmount: string; + status: StreamStatus; + updatedAt: number; + memo?: string; + endTimestamp?: number; +} + +export interface OptimisticMutation { + id: string; + streamId: string; + kind: MutationKind; + createdAt: number; + status: MutationStatus; + /** Field snapshot taken before applying `patch`, used to roll back. */ + before: Partial; + patch: Partial; + error?: string; +} + +export const STREAMS_STORE = 'cached_streams'; +export const MUTATIONS_STORE = 'optimistic_mutations'; + +const DB_NAME = 'flowfi-stream-cache'; +const DB_VERSION = 1; + +let dbPromise: Promise | null = null; + +export function isStorageAvailable(): boolean { + return typeof indexedDB !== 'undefined'; +} + +export function openStreamCacheDB(): Promise { + if (!isStorageAvailable()) { + return Promise.reject(new Error('IndexedDB is not available in this environment')); + } + if (dbPromise) return dbPromise; + + dbPromise = new Promise((resolve, reject) => { + const req = indexedDB.open(DB_NAME, DB_VERSION); + req.onupgradeneeded = () => { + const db = req.result; + if (!db.objectStoreNames.contains(STREAMS_STORE)) { + const store = db.createObjectStore(STREAMS_STORE, { keyPath: 'id' }); + store.createIndex('sender', 'sender'); + store.createIndex('recipient', 'recipient'); + store.createIndex('tokenAddress', 'tokenAddress'); + store.createIndex('updatedAt', 'updatedAt'); + } + if (!db.objectStoreNames.contains(MUTATIONS_STORE)) { + db.createObjectStore(MUTATIONS_STORE, { keyPath: 'id' }); + } + }; + req.onsuccess = () => resolve(req.result); + req.onerror = () => reject(req.error ?? new Error('Failed to open IndexedDB')); + }); + + return dbPromise; +} + +function reqToPromise(req: IDBRequest): Promise { + return new Promise((resolve, reject) => { + req.onsuccess = () => resolve(req.result); + req.onerror = () => reject(req.error ?? new Error('IndexedDB request failed')); + }); +} + +function txComplete(tx: IDBTransaction): Promise { + return new Promise((resolve, reject) => { + tx.oncomplete = () => resolve(); + tx.onerror = () => reject(tx.error ?? new Error('IndexedDB transaction failed')); + tx.onabort = () => reject(tx.error ?? new Error('IndexedDB transaction aborted')); + }); +} + +// --------------------------------------------------------------------------- +// Streams +// --------------------------------------------------------------------------- + +export async function getCachedStreams(): Promise { + const db = await openStreamCacheDB(); + const tx = db.transaction(STREAMS_STORE, 'readonly'); + return reqToPromise(tx.objectStore(STREAMS_STORE).getAll() as IDBRequest); +} + +export async function getCachedStream(id: string): Promise { + const db = await openStreamCacheDB(); + const tx = db.transaction(STREAMS_STORE, 'readonly'); + return reqToPromise( + tx.objectStore(STREAMS_STORE).get(id) as IDBRequest, + ); +} + +export async function putCachedStream(stream: CachedStream): Promise { + const db = await openStreamCacheDB(); + const tx = db.transaction(STREAMS_STORE, 'readwrite'); + tx.objectStore(STREAMS_STORE).put(stream); + await txComplete(tx); +} + +export async function putCachedStreams(streams: CachedStream[]): Promise { + if (streams.length === 0) return; + const db = await openStreamCacheDB(); + const tx = db.transaction(STREAMS_STORE, 'readwrite'); + const store = tx.objectStore(STREAMS_STORE); + for (const stream of streams) store.put(stream); + await txComplete(tx); +} + +export async function deleteCachedStream(id: string): Promise { + const db = await openStreamCacheDB(); + const tx = db.transaction(STREAMS_STORE, 'readwrite'); + tx.objectStore(STREAMS_STORE).delete(id); + await txComplete(tx); +} + +// --------------------------------------------------------------------------- +// Optimistic mutations +// --------------------------------------------------------------------------- + +export async function getOptimisticMutations(): Promise { + const db = await openStreamCacheDB(); + const tx = db.transaction(MUTATIONS_STORE, 'readonly'); + return reqToPromise( + tx.objectStore(MUTATIONS_STORE).getAll() as IDBRequest, + ); +} + +export async function putOptimisticMutation(mutation: OptimisticMutation): Promise { + const db = await openStreamCacheDB(); + const tx = db.transaction(MUTATIONS_STORE, 'readwrite'); + tx.objectStore(MUTATIONS_STORE).put(mutation); + await txComplete(tx); +} + +export async function deleteOptimisticMutation(id: string): Promise { + const db = await openStreamCacheDB(); + const tx = db.transaction(MUTATIONS_STORE, 'readwrite'); + tx.objectStore(MUTATIONS_STORE).delete(id); + await txComplete(tx); +} + +// --------------------------------------------------------------------------- +// Housekeeping +// --------------------------------------------------------------------------- + +export async function clearStreamCache(): Promise { + const db = await openStreamCacheDB(); + const tx = db.transaction([STREAMS_STORE, MUTATIONS_STORE], 'readwrite'); + tx.objectStore(STREAMS_STORE).clear(); + tx.objectStore(MUTATIONS_STORE).clear(); + await txComplete(tx); +}