Your Portfolio
-
Track your financing offers and returns
+
+ Track your financing offers and returns — updates stream in live
+ {lastUpdatedAt
+ ? · updated {relativeUpdate(lastUpdatedAt)}
+ : null}
+
- {offers.length > 0 && (
-
- {/* Extra earned stat */}
- {repaid.length > 0 && (
+ {/* Live earnings strip */}
+ {(active.length > 0 || repaid.length > 0) && (
-
-
- Est. yield earned: {totalEarned.toFixed(4)} (across {repaid.length} repaid offer{repaid.length !== 1 ? 's' : ''})
-
-
Based on agreed interest rates
+
+
+
+ Est. yield earned to date: ${totalEarnedToDateUsd.toFixed(2)}
+
+
Accruing in real time across {active.length} active position{active.length !== 1 ? 's' : ''}
+
+ {repaid.length > 0 && (
+
+
+ Realized yield: ${totalEarned.toFixed(2)} USD
+
+
Across {repaid.length} repaid offer{repaid.length !== 1 ? 's' : ''}
+
+ )}
)}
@@ -402,7 +499,7 @@ export default function PortfolioPage() {
{loading &&
}
{/* Empty state */}
- {!loading && offers.length === 0 && (
+ {!loading && positions.length === 0 && (
No financing offers yet.
@@ -416,50 +513,9 @@ export default function PortfolioPage() {
)}
- {offers.map(offer => {
- const Icon = STATUS_ICONS[offer.status] ?? Clock;
- return (
-
-
-
-
-
-
-
- {interestRateLabel(offer.interest_rate)} · {durationLabel(offer.duration)}
- {offer.funded_at > 0 && ` · Funded ${formatDate(offer.funded_at)}`}
-
- {(offer.status === 'Accepted' || offer.status === 'Financed') &&
- toStroopsBigInt(offer.amount_repaid) > 0n && (
-
- {formatAmount(toStroopsBigInt(offer.amount_repaid))} repaid ·{' '}
- {formatAmount(offerTotalDue(offer) - toStroopsBigInt(offer.amount_repaid))} remaining
-
- )}
-
-
-
-
- {formatAmount(offer.amount)} {offer.currency}
-
-
-
{offer.status}
-
- );
- })}
+ {positions.map(offer => (
+
+ ))}
{/* useSearchParams (the listing hand-off prefill) needs a Suspense boundary. */}
@@ -469,4 +525,4 @@ export default function PortfolioPage() {
);
-}
+}
\ No newline at end of file
diff --git a/invofi/apps/frontend/src/components/portfolio/ConnectionStatus.tsx b/invofi/apps/frontend/src/components/portfolio/ConnectionStatus.tsx
new file mode 100644
index 000000000..94ec553dd
--- /dev/null
+++ b/invofi/apps/frontend/src/components/portfolio/ConnectionStatus.tsx
@@ -0,0 +1,66 @@
+'use client';
+
+// ── Live connection status pill (issue #221) ─────────────────────────────────
+// Always-visible indicator of the dashboard's data source:
+// • green — streaming over WebSocket
+// • blue — polling fallback (Soroban events + Supabase resync)
+// • amber — connecting / reconnecting (with backoff detail)
+// • red — no live source available
+
+import { useLivePortfolio } from './LivePortfolioProvider';
+import type { ConnectionStatus } from '@/lib/live/types';
+import { cn } from '@/lib/utils';
+
+const STATUS_META: Record
= {
+ connected: {
+ label: 'Live · WebSocket',
+ dot: 'bg-green-500',
+ className:
+ 'border-green-200 bg-green-50 text-green-800 dark:border-green-900 dark:bg-green-950/40 dark:text-green-300',
+ },
+ polling: {
+ label: 'Live · Polling',
+ dot: 'bg-blue-500',
+ className:
+ 'border-blue-200 bg-blue-50 text-blue-800 dark:border-blue-900 dark:bg-blue-950/40 dark:text-blue-300',
+ },
+ connecting: {
+ label: 'Connecting…',
+ dot: 'bg-amber-500 animate-pulse',
+ className:
+ 'border-amber-200 bg-amber-50 text-amber-800 dark:border-amber-900 dark:bg-amber-950/40 dark:text-amber-300',
+ },
+ reconnecting: {
+ label: 'Reconnecting…',
+ dot: 'bg-amber-500 animate-pulse',
+ className:
+ 'border-amber-200 bg-amber-50 text-amber-800 dark:border-amber-900 dark:bg-amber-950/40 dark:text-amber-300',
+ },
+ offline: {
+ label: 'Offline',
+ dot: 'bg-red-500',
+ className:
+ 'border-red-200 bg-red-50 text-red-800 dark:border-red-900 dark:bg-red-950/40 dark:text-red-300',
+ },
+};
+
+export function ConnectionStatus() {
+ const { connection, connectionDetail } = useLivePortfolio();
+ const meta = STATUS_META[connection];
+
+ return (
+
+
+ {meta.label}
+ {connectionDetail && · {connectionDetail}}
+
+ );
+}
\ No newline at end of file
diff --git a/invofi/apps/frontend/src/components/portfolio/LivePortfolioProvider.tsx b/invofi/apps/frontend/src/components/portfolio/LivePortfolioProvider.tsx
new file mode 100644
index 000000000..e1a57973d
--- /dev/null
+++ b/invofi/apps/frontend/src/components/portfolio/LivePortfolioProvider.tsx
@@ -0,0 +1,179 @@
+'use client';
+
+// ── Live portfolio provider (issue #221) ─────────────────────────────────────
+// React context + useReducer that owns the live dashboard state. It loads the
+// lender's offers from the Supabase mirror, subscribes to the live engine
+// (WebSocket relay → Soroban event polling fallback), and re-derives every
+// position row (USD value, APY, earned-to-date, repayment progress) on each
+// update. Scoped to the portfolio route via `app/portfolio/layout.tsx`.
+
+import {
+ createContext,
+ useCallback,
+ useContext,
+ useEffect,
+ useMemo,
+ useReducer,
+ useRef,
+ useState,
+} from 'react';
+import type { FinancingOffer } from '@invofi/sdk';
+import { useWallet } from '@/components/auth/WalletProvider';
+import { supabase } from '@/lib/supabase';
+import { toStroopsBigInt } from '@/lib/utils';
+import { LivePortfolioEngine } from '@/lib/live/engine';
+import {
+ INITIAL_LIVE_PORTFOLIO_STATE,
+ livePortfolioReducer,
+} from '@/lib/live/reducer';
+import {
+ LIVE_CONTRACT_IDS,
+ LIVE_NETWORK_PASSPHRASE,
+ LIVE_RPC_URL,
+ LIVE_WS_URL,
+} from '@/lib/live/config';
+import { refreshXlmUsdPrice } from '@/lib/live/prices';
+import type { ConnectionStatus, LivePosition, LiveTransport } from '@/lib/live/types';
+
+export interface LivePortfolioContextValue {
+ positions: LivePosition[];
+ connection: ConnectionStatus;
+ connectionDetail: string | null;
+ transport: LiveTransport;
+ loading: boolean;
+ error: string | null;
+ lastUpdatedAt: number | null;
+ /** Force an immediate resync (also refreshes the XLM/USD price). */
+ refresh: () => void;
+}
+
+const LivePortfolioContext = createContext(null);
+
+export function LivePortfolioProvider({ children }: { children: React.ReactNode }) {
+ const { publicKey } = useWallet();
+ const [state, dispatch] = useReducer(livePortfolioReducer, INITIAL_LIVE_PORTFOLIO_STATE);
+ const engineRef = useRef(null);
+
+ // Restart the stream only when the authenticated user actually changes
+ // (wallet connect / sign-out), not on every Supabase token refresh.
+ const [sessionNonce, setSessionNonce] = useState(0);
+ useEffect(() => {
+ let cancelled = false;
+ let lastUserId: string | null = null;
+ void supabase.auth.getUser().then(({ data }) => {
+ if (cancelled) return;
+ lastUserId = data.user?.id ?? null;
+ });
+ const { data: { subscription } } = supabase.auth.onAuthStateChange(() => {
+ void supabase.auth.getUser().then(({ data }) => {
+ if (cancelled) return;
+ const id = data.user?.id ?? null;
+ if (id !== lastUserId) {
+ lastUserId = id;
+ setSessionNonce(nonce => nonce + 1);
+ }
+ });
+ });
+ return () => {
+ cancelled = true;
+ subscription.unsubscribe();
+ };
+ }, []);
+
+ /** Full-state fetch: the lender's offers from the Supabase mirror. */
+ const fetchPositions = useCallback(async (): Promise => {
+ const { data: { user } } = await supabase.auth.getUser();
+ if (!user) return [];
+ const { data, error } = await supabase
+ .from('financing_offers')
+ .select('*, invoice:invoices(*)')
+ .eq('lender_id', user.id)
+ .order('created_at', { ascending: false });
+ if (error) {
+ // Surface the failure in state (the engine swallows rejections), so the
+ // page shows an error banner instead of a misleading empty portfolio.
+ dispatch({ type: 'error', error: `Failed to load financing offers: ${error.message}` });
+ return [];
+ }
+ const rows = (data as unknown as FinancingOffer[]) ?? [];
+ // Normalize mirror strings to bigint stroops so math/display are consistent.
+ return rows.map(offer => ({
+ ...offer,
+ amount: toStroopsBigInt(offer.amount),
+ amount_repaid: toStroopsBigInt(offer.amount_repaid),
+ }));
+ }, []);
+
+ useEffect(() => {
+ let cancelled = false;
+ dispatch({ type: 'loading', loading: true });
+
+ const engine = new LivePortfolioEngine({
+ wsUrl: LIVE_WS_URL || null,
+ contractIds: LIVE_CONTRACT_IDS,
+ rpcUrl: LIVE_RPC_URL,
+ networkPassphrase: LIVE_NETWORK_PASSPHRASE,
+ fetchPositions,
+ onPositions: offers => {
+ if (!cancelled) dispatch({ type: 'positions', offers });
+ },
+ onUpdate: update => {
+ if (!cancelled) dispatch({ type: 'update', update });
+ },
+ onConnectionChange: (connection, transport, detail) => {
+ if (cancelled) return;
+ dispatch({ type: 'connection', connection, transport, detail });
+ },
+ });
+ engineRef.current = engine;
+
+ void engine.start().finally(() => {
+ if (!cancelled) dispatch({ type: 'loading', loading: false });
+ });
+
+ return () => {
+ cancelled = true;
+ engine.stop();
+ engineRef.current = null;
+ };
+ }, [fetchPositions, sessionNonce, publicKey]);
+
+ /** Refresh the XLM/USD price once on mount so USD values are real. */
+ useEffect(() => {
+ let cancelled = false;
+ void refreshXlmUsdPrice().then(() => {
+ if (!cancelled) engineRef.current?.resyncNow();
+ });
+ return () => {
+ cancelled = true;
+ };
+ }, []);
+
+ const refresh = useCallback(() => {
+ void refreshXlmUsdPrice().then(() => engineRef.current?.resyncNow());
+ }, []);
+
+ const value = useMemo(
+ () => ({
+ positions: state.positions,
+ connection: state.connection,
+ connectionDetail: state.connectionDetail,
+ transport: state.transport,
+ loading: state.loading,
+ error: state.error,
+ lastUpdatedAt: state.lastUpdatedAt,
+ refresh,
+ }),
+ [state, refresh],
+ );
+
+ return {children};
+}
+
+export function useLivePortfolio(): LivePortfolioContextValue {
+ const ctx = useContext(LivePortfolioContext);
+ if (!ctx) {
+ throw new Error('useLivePortfolio must be used within a LivePortfolioProvider');
+ }
+ return ctx;
+}
\ No newline at end of file
diff --git a/invofi/apps/frontend/src/components/portfolio/RepaymentProgress.tsx b/invofi/apps/frontend/src/components/portfolio/RepaymentProgress.tsx
new file mode 100644
index 000000000..9c68263a2
--- /dev/null
+++ b/invofi/apps/frontend/src/components/portfolio/RepaymentProgress.tsx
@@ -0,0 +1,37 @@
+'use client';
+
+// ── Streaming repayment progress bar (issue #221) ────────────────────────────
+// Fills as repayments stream in, animating the width transition so partial
+// repayments are visually obvious without a full refresh.
+
+import { cn } from '@/lib/utils';
+
+interface RepaymentProgressProps {
+ /** Ratio 0..1 of total due already repaid. */
+ value: number;
+ className?: string;
+ label?: string;
+}
+
+export function RepaymentProgress({ value, className, label = 'Repayment progress' }: RepaymentProgressProps) {
+ const clamped = Math.min(1, Math.max(0, value));
+ const pct = Math.round(clamped * 100);
+
+ return (
+
+ );
+}
\ No newline at end of file
diff --git a/invofi/apps/frontend/src/lib/live/config.test.ts b/invofi/apps/frontend/src/lib/live/config.test.ts
new file mode 100644
index 000000000..5211c1dd8
--- /dev/null
+++ b/invofi/apps/frontend/src/lib/live/config.test.ts
@@ -0,0 +1,47 @@
+import { afterEach, describe, expect, it, vi } from 'vitest';
+
+// Env is read at module load — reset the module registry so each test sees a
+// fresh process.env surface.
+
+vi.mock('@invofi/sdk', () => ({
+ Networks: {
+ TESTNET: 'Test SDF Network ; September 2015',
+ PUBLIC: 'Public Global Stellar Network ; September 2015',
+ },
+}));
+
+async function loadConfig() {
+ vi.resetModules();
+ return await import('./config');
+}
+
+afterEach(() => {
+ delete process.env.NEXT_PUBLIC_STELLAR_NETWORK;
+ delete process.env.NEXT_PUBLIC_RPC_URL;
+ delete process.env.NEXT_PUBLIC_WS_URL;
+});
+
+describe('live config', () => {
+ it('defaults to the testnet RPC on the testnet network', async () => {
+ process.env.NEXT_PUBLIC_STELLAR_NETWORK = 'testnet';
+ delete process.env.NEXT_PUBLIC_RPC_URL;
+ const { LIVE_RPC_URL, LIVE_NETWORK_PASSPHRASE } = await loadConfig();
+ expect(LIVE_RPC_URL).toBe('https://soroban-testnet.stellar.org');
+ expect(LIVE_NETWORK_PASSPHRASE).toContain('Test SDF Network');
+ });
+
+ it('defaults to the mainnet RPC when the network is mainnet', async () => {
+ process.env.NEXT_PUBLIC_STELLAR_NETWORK = 'mainnet';
+ delete process.env.NEXT_PUBLIC_RPC_URL;
+ const { LIVE_RPC_URL, LIVE_NETWORK_PASSPHRASE } = await loadConfig();
+ expect(LIVE_RPC_URL).toBe('https://soroban-rpc.stellar.org');
+ expect(LIVE_NETWORK_PASSPHRASE).toContain('Public Global Stellar Network');
+ });
+
+ it('honors an explicit RPC override regardless of network', async () => {
+ process.env.NEXT_PUBLIC_STELLAR_NETWORK = 'mainnet';
+ process.env.NEXT_PUBLIC_RPC_URL = 'https://rpc.example.com';
+ const { LIVE_RPC_URL } = await loadConfig();
+ expect(LIVE_RPC_URL).toBe('https://rpc.example.com');
+ });
+});
\ No newline at end of file
diff --git a/invofi/apps/frontend/src/lib/live/config.ts b/invofi/apps/frontend/src/lib/live/config.ts
new file mode 100644
index 000000000..e53aaaf07
--- /dev/null
+++ b/invofi/apps/frontend/src/lib/live/config.ts
@@ -0,0 +1,30 @@
+// ── Live dashboard config (issue #221) ───────────────────────────────────────
+// Reads the same env surface as `lib/contract.ts` so the live engine watches
+// the same contracts the app talks to. `NEXT_PUBLIC_WS_URL` is optional — when
+// unset the dashboard runs on the polling fallback.
+
+import { Networks } from '@invofi/sdk';
+
+export const LIVE_NETWORK = (process.env.NEXT_PUBLIC_STELLAR_NETWORK ?? 'testnet') as
+ | 'testnet'
+ | 'mainnet';
+
+/** Soroban RPC endpoint, derived from the network when not explicitly set. */
+export const LIVE_RPC_URL =
+ process.env.NEXT_PUBLIC_RPC_URL ??
+ (LIVE_NETWORK === 'mainnet'
+ ? 'https://soroban-rpc.stellar.org'
+ : 'https://soroban-testnet.stellar.org');
+
+export const LIVE_WS_URL = process.env.NEXT_PUBLIC_WS_URL ?? '';
+
+export const LIVE_NETWORK_PASSPHRASE =
+ LIVE_NETWORK === 'mainnet' ? Networks.PUBLIC : Networks.TESTNET;
+
+const LEGACY_CONTRACT_ID = process.env.NEXT_PUBLIC_CONTRACT_ID ?? '';
+const REGISTRY_ID = process.env.NEXT_PUBLIC_REGISTRY_CONTRACT_ID ?? LEGACY_CONTRACT_ID;
+const FINANCING_ID = process.env.NEXT_PUBLIC_FINANCING_CONTRACT_ID ?? LEGACY_CONTRACT_ID;
+const REPAYMENT_ID = process.env.NEXT_PUBLIC_REPAYMENT_CONTRACT_ID ?? LEGACY_CONTRACT_ID;
+
+/** Contract IDs to watch, deduped (legacy deployments route all three to one). */
+export const LIVE_CONTRACT_IDS = [...new Set([REGISTRY_ID, FINANCING_ID, REPAYMENT_ID])].filter(Boolean);
\ No newline at end of file
diff --git a/invofi/apps/frontend/src/lib/live/convert.test.ts b/invofi/apps/frontend/src/lib/live/convert.test.ts
new file mode 100644
index 000000000..1f2cb4ef1
--- /dev/null
+++ b/invofi/apps/frontend/src/lib/live/convert.test.ts
@@ -0,0 +1,40 @@
+import { describe, expect, it } from 'vitest';
+import { safeStroopsFromWire, stroopsFromWire } from './convert';
+
+describe('stroopsFromWire', () => {
+ it('passes bigints and numbers through as stroops', () => {
+ expect(stroopsFromWire(1_000_000n)).toBe(1_000_000n);
+ expect(stroopsFromWire(1_000_000)).toBe(1_000_000n);
+ });
+
+ it('rejects unsafe numeric amounts instead of storing a rounded value', () => {
+ // 9007199254740993 is above Number.MAX_SAFE_INTEGER — a JSON number this
+ // large is already rounded before this function sees it.
+ expect(() => stroopsFromWire(9_007_199_254_740_993)).toThrow(RangeError);
+ expect(() => stroopsFromWire(1.5)).toThrow(RangeError);
+ expect(() => stroopsFromWire(NaN)).toThrow(RangeError);
+ });
+
+ it('treats integer strings as raw stroops (wire convention)', () => {
+ expect(stroopsFromWire('1000000')).toBe(1_000_000n);
+ expect(stroopsFromWire('-500')).toBe(-500n);
+ });
+
+ it('treats decimal strings as human units, unlike integer strings', () => {
+ expect(stroopsFromWire('1.0')).toBe(10_000_000n);
+ expect(stroopsFromWire('1.5')).toBe(15_000_000n);
+ });
+
+ it('normalizes empty and nullish values', () => {
+ expect(stroopsFromWire('')).toBe(0n);
+ expect(stroopsFromWire(null)).toBe(0n);
+ expect(stroopsFromWire(undefined)).toBe(0n);
+ });
+
+ it('safeStroopsFromWire never throws on malformed values', () => {
+ expect(safeStroopsFromWire(9_007_199_254_740_993)).toBe(0n);
+ expect(safeStroopsFromWire({} as never)).toBe(0n);
+ expect(safeStroopsFromWire('not-a-number')).toBe(0n);
+ expect(safeStroopsFromWire(1_000_000)).toBe(1_000_000n);
+ });
+});
\ No newline at end of file
diff --git a/invofi/apps/frontend/src/lib/live/convert.ts b/invofi/apps/frontend/src/lib/live/convert.ts
new file mode 100644
index 000000000..e2c80846d
--- /dev/null
+++ b/invofi/apps/frontend/src/lib/live/convert.ts
@@ -0,0 +1,38 @@
+// ── Wire-amount conversion (issue #221) ──────────────────────────────────────
+// BigInts do not survive JSON, so relay messages carry amounts as integer
+// strings that are already in stroops (on-chain i128 values). The general
+// `toStroopsBigInt` from `lib/utils` interprets integer strings as *human
+// units* (used by the Supabase mirror), which would silently inflate a
+// streamed stroop amount by 10⁷. This helper converts wire values explicitly:
+// integer strings/numbers → stroops as-is; decimal strings → human units.
+
+import { toStroopsBigInt } from '@/lib/utils';
+
+export function stroopsFromWire(value: bigint | number | string | null | undefined): bigint {
+ if (value === null || value === undefined) return 0n;
+ if (typeof value === 'bigint') return value;
+ if (typeof value === 'number') {
+ // A JSON number above Number.MAX_SAFE_INTEGER is already rounded before
+ // this function sees it — never store a silently-rounded stroop amount.
+ // Large wire amounts must arrive as integer strings (on-chain i128).
+ if (!Number.isSafeInteger(value)) {
+ throw new RangeError(`Unsafe numeric wire amount: ${value}`);
+ }
+ return BigInt(value);
+ }
+ const s = String(value).trim();
+ if (s === '') return 0n;
+ // Integer string → already stroops (wire convention).
+ if (/^-?\d+$/.test(s)) return BigInt(s);
+ // Decimal string → human units from the mirror; convert.
+ return toStroopsBigInt(s);
+}
+
+/** Like {@link stroopsFromWire} but never throws — malformed values become 0n. */
+export function safeStroopsFromWire(value: bigint | number | string | null | undefined): bigint {
+ try {
+ return stroopsFromWire(value);
+ } catch {
+ return 0n;
+ }
+}
\ No newline at end of file
diff --git a/invofi/apps/frontend/src/lib/live/engine.test.ts b/invofi/apps/frontend/src/lib/live/engine.test.ts
new file mode 100644
index 000000000..93c859503
--- /dev/null
+++ b/invofi/apps/frontend/src/lib/live/engine.test.ts
@@ -0,0 +1,236 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import type { FinancingOffer } from '@invofi/sdk';
+import { LivePortfolioEngine } from './engine';
+import type { LivePositionUpdate } from './types';
+
+const {
+ wsStartMock,
+ wsStopMock,
+ pollingStartMock,
+ pollingStopMock,
+ createWsMock,
+ createPollingMock,
+} = vi.hoisted(() => ({
+ wsStartMock: vi.fn(),
+ wsStopMock: vi.fn(),
+ pollingStartMock: vi.fn(),
+ pollingStopMock: vi.fn(),
+ createWsMock: vi.fn(),
+ createPollingMock: vi.fn(),
+}));
+
+vi.mock('./transports', () => ({
+ createWebSocketTransport: (opts: unknown) => {
+ createWsMock(opts);
+ return { start: wsStartMock, stop: wsStopMock };
+ },
+ createPollingTransport: (opts: unknown) => {
+ createPollingMock(opts);
+ return { start: pollingStartMock, stop: pollingStopMock };
+ },
+}));
+
+/** Funded "now" so accrual is partial and observable across ticks. */
+const makeActiveOffer = (nowSecs: number): FinancingOffer => ({
+ id: 'off_1',
+ invoice_id: 'inv_1',
+ lender: 'GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAWHF',
+ amount: 10_000_000n,
+ currency: 'USDC',
+ interest_rate: 500,
+ duration: 40,
+ amount_repaid: 0n,
+ status: 'Financed',
+ funded_at: nowSecs,
+});
+
+describe('LivePortfolioEngine', () => {
+ beforeEach(() => {
+ vi.useFakeTimers();
+ wsStartMock.mockClear();
+ wsStopMock.mockClear();
+ pollingStartMock.mockClear();
+ pollingStopMock.mockClear();
+ createWsMock.mockClear();
+ createPollingMock.mockClear();
+ });
+
+ afterEach(() => {
+ vi.useRealTimers();
+ });
+
+ it('degrades to the polling transport when no WebSocket relay is configured', async () => {
+ const statuses: Array<[string, string]> = [];
+ const engine = new LivePortfolioEngine({
+ wsUrl: null,
+ contractIds: ['registry'],
+ rpcUrl: 'https://rpc',
+ networkPassphrase: 'testnet',
+ fetchPositions: async () => [makeActiveOffer(Date.now() / 1000)],
+ onPositions: () => {},
+ onUpdate: () => {},
+ onConnectionChange: (connection, transport) => statuses.push([connection, transport]),
+ });
+
+ await engine.start();
+
+ expect(wsStartMock).not.toHaveBeenCalled();
+ expect(pollingStartMock).toHaveBeenCalledTimes(1);
+ expect(statuses).toContainEqual(['polling', 'polling']);
+ // Contract + network wiring reaches the polling transport.
+ const pollingOptions = createPollingMock.mock.calls[0][0] as Record;
+ expect(pollingOptions.contractIds).toEqual(['registry']);
+ expect(pollingOptions.rpcUrl).toBe('https://rpc');
+ expect(pollingOptions.networkPassphrase).toBe('testnet');
+ engine.stop();
+ });
+
+ it('prefers the WebSocket relay when configured', async () => {
+ const engine = new LivePortfolioEngine({
+ wsUrl: 'wss://relay.invofi.dev',
+ contractIds: ['registry'],
+ rpcUrl: 'https://rpc',
+ networkPassphrase: 'testnet',
+ fetchPositions: async () => [],
+ onPositions: () => {},
+ onUpdate: () => {},
+ onConnectionChange: () => {},
+ });
+
+ await engine.start();
+
+ expect(wsStartMock).toHaveBeenCalledTimes(1);
+ expect(pollingStartMock).not.toHaveBeenCalled();
+ // URL and callbacks reach the WebSocket transport.
+ const wsOptions = createWsMock.mock.calls[0][0] as Record;
+ expect(wsOptions.url).toBe('wss://relay.invofi.dev');
+ expect(typeof wsOptions.onUpdate).toBe('function');
+ expect(typeof wsOptions.onGiveUp).toBe('function');
+ engine.stop();
+ });
+
+ it('degrades to polling when the configured relay gives up', async () => {
+ const statuses: Array<[string, string]> = [];
+ const engine = new LivePortfolioEngine({
+ wsUrl: 'wss://relay.invofi.dev',
+ contractIds: ['registry'],
+ rpcUrl: 'https://rpc',
+ networkPassphrase: 'testnet',
+ fetchPositions: async () => [],
+ onPositions: () => {},
+ onUpdate: () => {},
+ onConnectionChange: (connection, transport) => statuses.push([connection, transport]),
+ });
+
+ await engine.start();
+ const wsOptions = createWsMock.mock.calls[0][0] as { onGiveUp: (reason: string) => void };
+ wsOptions.onGiveUp('relay unreachable');
+
+ expect(wsStopMock).toHaveBeenCalledTimes(1);
+ expect(pollingStartMock).toHaveBeenCalledTimes(1);
+ expect(statuses).toContainEqual(['polling', 'polling']);
+ engine.stop();
+ });
+
+ it('accrues yield in real time, throttled to one update per position per second', async () => {
+ const updates: Array<{ kind: string; positionId: string; earnedToDate?: bigint }> = [];
+ const offer = makeActiveOffer(Date.now() / 1000);
+ const engine = new LivePortfolioEngine({
+ wsUrl: null,
+ contractIds: [],
+ rpcUrl: 'https://rpc',
+ networkPassphrase: 'testnet',
+ fetchPositions: async () => [offer],
+ onPositions: () => {},
+ onUpdate: update => updates.push(update),
+ onConnectionChange: () => {},
+ throttleMs: 1_000,
+ yieldTickMs: 250,
+ });
+
+ await engine.start();
+ expect(updates).toHaveLength(0);
+
+ // First accrual lands at 250ms (leading edge); the coalescing window
+ // closes at +1000ms delivering the latest accrued value.
+ await vi.advanceTimersByTimeAsync(1_300);
+ expect(updates).toHaveLength(2);
+ expect(updates[0].kind).toBe('yield_calculated');
+ expect(updates[0].positionId).toBe('off_1');
+ // Accrual advances with the clock, not just the tick count.
+ expect(updates[1].earnedToDate!).toBeGreaterThan(updates[0].earnedToDate!);
+
+ // One more second → exactly one more delivery (never more than 1/sec).
+ await vi.advanceTimersByTimeAsync(1_300);
+ expect(updates).toHaveLength(3);
+ expect(updates[2].earnedToDate!).toBeGreaterThan(updates[1].earnedToDate!);
+
+ engine.stop();
+ });
+
+ it('delivers repayment and yield updates independently for the same position', async () => {
+ const updates: LivePositionUpdate[] = [];
+ const engine = new LivePortfolioEngine({
+ wsUrl: 'wss://relay.invofi.dev',
+ contractIds: ['registry'],
+ rpcUrl: 'https://rpc',
+ networkPassphrase: 'testnet',
+ fetchPositions: async () => [makeActiveOffer(Date.now() / 1000)],
+ onPositions: () => {},
+ onUpdate: update => updates.push(update),
+ onConnectionChange: () => {},
+ throttleMs: 1_000,
+ yieldTickMs: 250,
+ });
+
+ await engine.start();
+
+ const wsOptions = createWsMock.mock.calls[0][0] as {
+ onUpdate: (update: LivePositionUpdate) => void;
+ };
+ // A repayment arrives while the yield timer is running.
+ wsOptions.onUpdate({
+ kind: 'repayment_received',
+ positionId: 'off_1',
+ amountRepaid: 1_000_000n,
+ fullyRepaid: false,
+ });
+
+ await vi.advanceTimersByTimeAsync(1_300);
+
+ const kinds = updates.map(u => u.kind);
+ expect(kinds).toContain('repayment_received');
+ expect(kinds).toContain('yield_calculated');
+ engine.stop();
+ });
+
+ it('stop() halts yield ticks and resyncs', async () => {
+ const updates: unknown[] = [];
+ let fetches = 0;
+ const engine = new LivePortfolioEngine({
+ wsUrl: null,
+ contractIds: [],
+ rpcUrl: 'https://rpc',
+ networkPassphrase: 'testnet',
+ fetchPositions: async () => {
+ fetches++;
+ return [makeActiveOffer(Date.now() / 1000)];
+ },
+ onPositions: () => {},
+ onUpdate: update => updates.push(update),
+ onConnectionChange: () => {},
+ resyncIntervalMs: 500,
+ yieldTickMs: 250,
+ });
+
+ await engine.start();
+ engine.stop();
+ const fetchesAtStop = fetches;
+
+ await vi.advanceTimersByTimeAsync(5_000);
+
+ expect(updates).toHaveLength(0);
+ expect(fetches).toBe(fetchesAtStop);
+ expect(pollingStopMock).toHaveBeenCalledTimes(1);
+ });
+});
\ No newline at end of file
diff --git a/invofi/apps/frontend/src/lib/live/engine.ts b/invofi/apps/frontend/src/lib/live/engine.ts
new file mode 100644
index 000000000..cd9a7daaa
--- /dev/null
+++ b/invofi/apps/frontend/src/lib/live/engine.ts
@@ -0,0 +1,198 @@
+// ── Live portfolio engine (issue #221) ───────────────────────────────────────
+// Owns the transport lifecycle and the per-position throttling:
+//
+// 1. Resyncs positions from the caller's fetch function (initial + periodic +
+// reconnect safety net).
+// 2. Prefers a WebSocket relay; degrades to Soroban event polling when the
+// relay is missing or unreachable (graceful-degradation criterion).
+// 3. Recomputes accruing yield once per second for active positions, so
+// "earned to date" moves in real time even between discrete events.
+// 4. Routes every update through a per-position throttle (≤ 1/sec/position)
+// before handing it to the reducer.
+
+import type { FinancingOffer } from '@invofi/sdk';
+import type { PerKeyThrottle } from './throttle';
+import { createPerKeyThrottle } from './throttle';
+import type { ConnectionStatus, LivePositionUpdate, LiveTransport } from './types';
+import { offerApy, yieldEarnedStroops, isActiveOffer } from './yield';
+import { createPollingTransport, createWebSocketTransport, type TransportHandle } from './transports';
+
+export interface LivePortfolioEngineOptions {
+ wsUrl: string | null;
+ contractIds: string[];
+ rpcUrl: string;
+ networkPassphrase: string;
+ /** Full-state fetch (e.g. Supabase mirror). Never rejects — return [] on failure. */
+ fetchPositions: () => Promise;
+ onPositions: (offers: FinancingOffer[]) => void;
+ onUpdate: (update: LivePositionUpdate) => void;
+ onConnectionChange: (status: ConnectionStatus, transport: LiveTransport, detail?: string) => void;
+ /** Max one delivered update per position per this window. Default 1000ms. */
+ throttleMs?: number;
+ /** Periodic full resync in both modes. Default 10s. */
+ resyncIntervalMs?: number;
+ /** How often to recompute accruing yield. Default 1000ms. */
+ yieldTickMs?: number;
+}
+
+export class LivePortfolioEngine {
+ private readonly opts: LivePortfolioEngineOptions;
+ private ws: TransportHandle | null = null;
+ private polling: TransportHandle | null = null;
+ private throttle: PerKeyThrottle | null = null;
+ private resyncTimer: ReturnType | null = null;
+ private yieldTimer: ReturnType | null = null;
+ private latestOffers: FinancingOffer[] = [];
+ /** Last dispatched accrual per position — unchanged ticks are skipped. */
+ private readonly lastEarned = new Map();
+ private started = false;
+ private stopped = false;
+ private degradedToPolling = false;
+ private resyncInFlight: Promise | null = null;
+ private resyncGeneration = 0;
+
+ constructor(options: LivePortfolioEngineOptions) {
+ this.opts = options;
+ }
+
+ /** Throttle key: one coalescing slot per position per update kind. */
+ private static key(update: LivePositionUpdate): string {
+ return `${update.positionId}:${update.kind}`;
+ }
+
+ /** Establish the live stream. Resolves after the first positions resync. */
+ async start(): Promise {
+ if (this.started) return;
+ this.started = true;
+
+ const {
+ wsUrl,
+ contractIds,
+ rpcUrl,
+ networkPassphrase,
+ onUpdate,
+ onConnectionChange,
+ throttleMs = 1_000,
+ resyncIntervalMs = 10_000,
+ yieldTickMs = 1_000,
+ } = this.opts;
+
+ this.throttle = createPerKeyThrottle(throttleMs, (_key, update) => {
+ onUpdate(update);
+ });
+
+ if (wsUrl) {
+ this.ws = createWebSocketTransport({
+ url: wsUrl,
+ onUpdate: update => this.throttle?.(LivePortfolioEngine.key(update), update),
+ onConnectionChange: (status, detail) => onConnectionChange(status, 'websocket', detail),
+ onResync: () => void this.resync(),
+ onGiveUp: reason => this.degradeToPolling(reason),
+ });
+ this.ws.start();
+ } else {
+ this.degradeToPolling('no WebSocket relay configured');
+ }
+
+ // Continuous accrual: recompute yields for active positions each tick.
+ // Unchanged values are skipped — a fresh updatedAt alone isn't an update.
+ this.yieldTimer = setInterval(() => {
+ if (this.stopped) return;
+ const nowSecs = Date.now() / 1000;
+ for (const offer of this.latestOffers) {
+ if (!isActiveOffer(offer)) continue;
+ const earnedToDate = yieldEarnedStroops(offer, nowSecs);
+ if (this.lastEarned.get(offer.id) === earnedToDate) continue;
+ this.lastEarned.set(offer.id, earnedToDate);
+ const update: LivePositionUpdate = {
+ kind: 'yield_calculated',
+ positionId: offer.id,
+ apy: offerApy(offer),
+ earnedToDate,
+ updatedAt: Date.now(),
+ };
+ this.throttle?.(LivePortfolioEngine.key(update), update);
+ }
+ }, yieldTickMs);
+
+ // Periodic safety-net resync (both modes) so brand-new positions appear.
+ this.resyncTimer = setInterval(() => void this.resync(), resyncIntervalMs);
+
+ await this.resync();
+ }
+
+ /** Switch from the relay to the polling fallback, once. */
+ private degradeToPolling(reason: string): void {
+ if (this.stopped || this.degradedToPolling) return;
+ this.degradedToPolling = true;
+ this.ws?.stop();
+ this.ws = null;
+ this.opts.onConnectionChange('polling', 'polling', reason);
+ this.polling = createPollingTransport({
+ contractIds: this.opts.contractIds,
+ rpcUrl: this.opts.rpcUrl,
+ networkPassphrase: this.opts.networkPassphrase,
+ onUpdate: update => this.throttle?.(LivePortfolioEngine.key(update), update),
+ onConnectionChange: (status, detail) =>
+ this.opts.onConnectionChange(status, 'polling', detail),
+ onResync: () => void this.resync(),
+ });
+ this.polling.start();
+ }
+
+ /**
+ * Full-state resync. In-flight requests are shared (never concurrent), and a
+ * generation counter discards responses superseded by a newer resync so a
+ * stale response can't overwrite fresher positions.
+ */
+ private resync(): Promise {
+ if (this.stopped) return Promise.resolve();
+ if (this.resyncInFlight) return this.resyncInFlight;
+ const generation = ++this.resyncGeneration;
+ this.resyncInFlight = this.opts
+ .fetchPositions()
+ .then(offers => {
+ if (this.stopped || generation !== this.resyncGeneration) return;
+ this.latestOffers = offers;
+ this.opts.onPositions(offers);
+ })
+ .catch(() => {
+ // A failed resync is non-fatal — the live stream keeps the last state.
+ })
+ .then(() => {
+ if (generation === this.resyncGeneration) this.resyncInFlight = null;
+ });
+ return this.resyncInFlight;
+ }
+
+ /** Force an immediate full resync (used by refresh buttons + auth changes). */
+ resyncNow(): void {
+ void this.resync();
+ }
+
+ /**
+ * Tear down every timer and transport. Terminal: an engine cannot be
+ * restarted after `stop()`. Construct a new instance instead.
+ */
+ stop(): void {
+ this.stopped = true;
+ this.resyncGeneration += 1;
+ if (this.resyncTimer) {
+ clearInterval(this.resyncTimer);
+ this.resyncTimer = null;
+ }
+ if (this.yieldTimer) {
+ clearInterval(this.yieldTimer);
+ this.yieldTimer = null;
+ }
+ this.ws?.stop();
+ this.ws = null;
+ this.polling?.stop();
+ this.polling = null;
+ // Deliver anything still pending before dropping the throttle.
+ this.throttle?.flush();
+ this.throttle?.stop();
+ this.throttle = null;
+ this.lastEarned.clear();
+ }
+}
\ No newline at end of file
diff --git a/invofi/apps/frontend/src/lib/live/prices.test.ts b/invofi/apps/frontend/src/lib/live/prices.test.ts
new file mode 100644
index 000000000..3f54d85da
--- /dev/null
+++ b/invofi/apps/frontend/src/lib/live/prices.test.ts
@@ -0,0 +1,37 @@
+import { afterEach, describe, expect, it, vi } from 'vitest';
+import { __setXlmUsdPriceForTests, refreshXlmUsdPrice, stroopsToUsd, usdPriceFor } from './prices';
+
+describe('prices', () => {
+ afterEach(() => {
+ __setXlmUsdPriceForTests(null);
+ vi.unstubAllGlobals();
+ });
+
+ it('treats USDC as a $1 stablecoin', () => {
+ expect(usdPriceFor('USDC')).toBe(1);
+ });
+
+ it('uses the cached XLM price once set, defaulting to $1 otherwise', () => {
+ __setXlmUsdPriceForTests(0.5);
+ expect(usdPriceFor('XLM')).toBe(0.5);
+ expect(stroopsToUsd(10_000_000n, 'XLM')).toBeCloseTo(0.5, 5);
+
+ __setXlmUsdPriceForTests(null);
+ expect(usdPriceFor('XLM')).toBe(1);
+ });
+
+ it('falls back without rejecting when the live price feed is unreachable', async () => {
+ vi.stubGlobal('fetch', vi.fn(() => Promise.reject(new Error('offline'))));
+ await expect(refreshXlmUsdPrice()).resolves.toBeGreaterThan(0);
+ expect(usdPriceFor('XLM')).toBeGreaterThan(0);
+ });
+
+ it('uses a fetched price when the feed responds', async () => {
+ vi.stubGlobal(
+ 'fetch',
+ vi.fn(async () => new Response(JSON.stringify({ stellar: { usd: 0.42 } }), { status: 200 })),
+ );
+ await expect(refreshXlmUsdPrice()).resolves.toBeCloseTo(0.42, 5);
+ expect(usdPriceFor('XLM')).toBeCloseTo(0.42, 5);
+ });
+});
\ No newline at end of file
diff --git a/invofi/apps/frontend/src/lib/live/prices.ts b/invofi/apps/frontend/src/lib/live/prices.ts
new file mode 100644
index 000000000..1cd6c2a80
--- /dev/null
+++ b/invofi/apps/frontend/src/lib/live/prices.ts
@@ -0,0 +1,72 @@
+// ── USD pricing (issue #221) ─────────────────────────────────────────────────
+// The live dashboard shows every position value in USD. USDC is a stablecoin
+// (peg ≈ $1). XLM is priced from a light, cached source: an env override wins,
+// otherwise CoinGecko's public endpoint is tried, and both fall back to the
+// override's default so the dashboard still renders offline.
+
+import type { Currency } from '@invofi/sdk';
+import { STROOPS_PER_XLM } from '@/lib/constants';
+
+const XLM_USD_CACHE_TTL_MS = 5 * 60_000;
+
+const ENV_XLM_USD_PRICE = Number(process.env.NEXT_PUBLIC_XLM_USD_PRICE ?? '');
+const DEFAULT_XLM_USD_PRICE = 1;
+
+let cachedXlmUsd: number | null = null;
+let cachedXlmUsdAt = 0;
+
+/** Best known XLM price, or the fallback when nothing is cached yet. */
+function xlmUsdPrice(): number {
+ if (cachedXlmUsd !== null) return cachedXlmUsd;
+ if (Number.isFinite(ENV_XLM_USD_PRICE) && ENV_XLM_USD_PRICE > 0) return ENV_XLM_USD_PRICE;
+ return DEFAULT_XLM_USD_PRICE;
+}
+
+/**
+ * USD price for a currency. USDC is treated as $1; XLM uses the cached price
+ * (see {@link refreshXlmUsdPrice}).
+ */
+export function usdPriceFor(currency: Currency): number {
+ return currency === 'USDC' ? 1 : xlmUsdPrice();
+}
+
+/** Convert stroops to USD at the current price. */
+export function stroopsToUsd(stroops: bigint, currency: Currency): number {
+ return (Number(stroops) / STROOPS_PER_XLM) * usdPriceFor(currency);
+}
+
+/**
+ * Refresh the cached XLM/USD price. Tries the CoinGecko public endpoint and
+ * falls back to the `NEXT_PUBLIC_XLM_USD_PRICE` env override, then the
+ * built-in default. Never rejects — a failure just keeps the last price.
+ */
+export async function refreshXlmUsdPrice(): Promise {
+ if (cachedXlmUsd !== null && Date.now() - cachedXlmUsdAt < XLM_USD_CACHE_TTL_MS) {
+ return cachedXlmUsd;
+ }
+ try {
+ const res = await fetch(
+ 'https://api.coingecko.com/api/v3/simple/price?ids=stellar&vs_currencies=usd',
+ );
+ if (res.ok) {
+ const json = (await res.json()) as { stellar?: { usd?: number } };
+ const price = Number(json?.stellar?.usd);
+ if (Number.isFinite(price) && price > 0) {
+ cachedXlmUsd = price;
+ cachedXlmUsdAt = Date.now();
+ return price;
+ }
+ }
+ } catch {
+ // Network unavailable — fall through to the override / default.
+ }
+ cachedXlmUsd = xlmUsdPrice();
+ cachedXlmUsdAt = Date.now();
+ return cachedXlmUsd;
+}
+
+/** Test hook — pin the cached XLM price without a network call. */
+export function __setXlmUsdPriceForTests(price: number | null): void {
+ cachedXlmUsd = price;
+ cachedXlmUsdAt = price === null ? 0 : Date.now();
+}
\ No newline at end of file
diff --git a/invofi/apps/frontend/src/lib/live/reducer.test.ts b/invofi/apps/frontend/src/lib/live/reducer.test.ts
new file mode 100644
index 000000000..f6a28f1cd
--- /dev/null
+++ b/invofi/apps/frontend/src/lib/live/reducer.test.ts
@@ -0,0 +1,259 @@
+import { describe, expect, it } from 'vitest';
+import type { FinancingOffer } from '@invofi/sdk';
+import {
+ INITIAL_LIVE_PORTFOLIO_STATE,
+ deriveLivePosition,
+ livePortfolioReducer,
+} from './reducer';
+import { __setXlmUsdPriceForTests } from './prices';
+
+const DAY = 86_400;
+const offer: FinancingOffer = {
+ id: 'off_1',
+ invoice_id: 'inv_1',
+ lender: 'GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAWHF',
+ amount: 10_000_000n, // 1 XLM
+ currency: 'USDC',
+ interest_rate: 500,
+ duration: 30 * DAY,
+ amount_repaid: 0n,
+ status: 'Financed',
+ funded_at: 1_000_000,
+};
+
+describe('deriveLivePosition', () => {
+ it('computes apy, earned-to-date, remaining, progress, and USD value', () => {
+ __setXlmUsdPriceForTests(null); // XLM fallback = 1
+ const live = deriveLivePosition(offer, 1_000_000 + 15 * DAY);
+
+ expect(live.apy).toBeCloseTo(60.83, 1);
+ expect(live.earnedToDate).toBe(250_000n); // half of the 5% yield
+ expect(live.totalDue).toBe(10_500_000n);
+ expect(live.remaining).toBe(10_500_000n);
+ expect(live.repaymentProgress).toBe(0);
+ expect(live.liveValueUsd).toBeCloseTo(1.05, 5);
+ });
+});
+
+describe('livePortfolioReducer', () => {
+ it('starts loading with an empty portfolio', () => {
+ expect(livePortfolioReducer(INITIAL_LIVE_PORTFOLIO_STATE, { type: 'reset' })).toEqual(
+ INITIAL_LIVE_PORTFOLIO_STATE,
+ );
+ });
+
+ it('derives live rows from a full positions resync', () => {
+ const state = livePortfolioReducer(INITIAL_LIVE_PORTFOLIO_STATE, {
+ type: 'positions',
+ offers: [offer],
+ });
+
+ expect(state.loading).toBe(false);
+ expect(state.positions).toHaveLength(1);
+ expect(state.positions[0].remaining).toBe(10_500_000n);
+ expect(state.lastUpdatedAt).not.toBeNull();
+ });
+
+ it('applies a streamed repayment to the matching position', () => {
+ let state = livePortfolioReducer(INITIAL_LIVE_PORTFOLIO_STATE, {
+ type: 'positions',
+ offers: [offer],
+ });
+
+ state = livePortfolioReducer(state, {
+ type: 'update',
+ update: {
+ kind: 'repayment_received',
+ positionId: 'off_1',
+ amountRepaid: 2_000_000n,
+ fullyRepaid: false,
+ updatedAt: 1_000,
+ },
+ });
+
+ expect(state.positions[0].amount_repaid).toBe(2_000_000n);
+ expect(state.positions[0].remaining).toBe(8_500_000n);
+ expect(state.positions[0].repaymentProgress).toBeCloseTo(2_000_000 / 10_500_000, 5);
+ });
+
+ it('marks a position Repaid when the final repayment arrives', () => {
+ let state = livePortfolioReducer(INITIAL_LIVE_PORTFOLIO_STATE, {
+ type: 'positions',
+ offers: [offer],
+ });
+
+ state = livePortfolioReducer(state, {
+ type: 'update',
+ update: {
+ kind: 'repayment_received',
+ positionId: 'off_1',
+ amountRepaid: 10_500_000n,
+ fullyRepaid: true,
+ },
+ });
+
+ expect(state.positions[0].status).toBe('Repaid');
+ expect(state.positions[0].repaymentProgress).toBe(1);
+ });
+
+ it('uses a cumulative remaining to keep replayed repayments idempotent', () => {
+ let state = livePortfolioReducer(INITIAL_LIVE_PORTFOLIO_STATE, {
+ type: 'positions',
+ offers: [offer],
+ });
+
+ const replay = {
+ kind: 'repayment_received' as const,
+ positionId: 'off_1',
+ amountRepaid: 2_000_000n,
+ remaining: 8_500_000n,
+ fullyRepaid: false,
+ };
+
+ state = livePortfolioReducer(state, { type: 'update', update: replay });
+ expect(state.positions[0].amount_repaid).toBe(2_000_000n);
+ expect(state.positions[0].remaining).toBe(8_500_000n);
+
+ // The same delivery replays after a reconnect — must not double-count.
+ state = livePortfolioReducer(state, { type: 'update', update: replay });
+ expect(state.positions[0].amount_repaid).toBe(2_000_000n);
+ expect(state.positions[0].remaining).toBe(8_500_000n);
+ });
+
+ it('ignores stale cumulative repayments that arrived out of order', () => {
+ let state = livePortfolioReducer(INITIAL_LIVE_PORTFOLIO_STATE, {
+ type: 'positions',
+ offers: [offer],
+ });
+
+ // A fresh delivery first...
+ state = livePortfolioReducer(state, {
+ type: 'update',
+ update: {
+ kind: 'repayment_received',
+ positionId: 'off_1',
+ amountRepaid: 2_000_000n,
+ remaining: 8_500_000n,
+ fullyRepaid: false,
+ },
+ });
+
+ // ...then an older delivery replays with a higher (staler) remaining.
+ state = livePortfolioReducer(state, {
+ type: 'update',
+ update: {
+ kind: 'repayment_received',
+ positionId: 'off_1',
+ amountRepaid: 2_000_000n,
+ remaining: 9_500_000n,
+ fullyRepaid: false,
+ },
+ });
+
+ expect(state.positions[0].amount_repaid).toBe(2_000_000n);
+ expect(state.positions[0].remaining).toBe(8_500_000n);
+ });
+
+ it('falls back to incremental amounts when no cumulative value is sent', () => {
+ let state = livePortfolioReducer(INITIAL_LIVE_PORTFOLIO_STATE, {
+ type: 'positions',
+ offers: [offer],
+ });
+
+ state = livePortfolioReducer(state, {
+ type: 'update',
+ update: {
+ kind: 'repayment_received',
+ positionId: 'off_1',
+ amountRepaid: 2_000_000n,
+ fullyRepaid: false,
+ },
+ });
+
+ expect(state.positions[0].amount_repaid).toBe(2_000_000n);
+ });
+
+ it('applies a yield_calculated stream update', () => {
+ let state = livePortfolioReducer(INITIAL_LIVE_PORTFOLIO_STATE, {
+ type: 'positions',
+ offers: [offer],
+ });
+
+ state = livePortfolioReducer(state, {
+ type: 'update',
+ update: {
+ kind: 'yield_calculated',
+ positionId: 'off_1',
+ apy: 61,
+ earnedToDate: 300_000n,
+ },
+ });
+
+ expect(state.positions[0].apy).toBe(61);
+ expect(state.positions[0].earnedToDate).toBe(300_000n);
+ });
+
+ it('merges partial offer fields from a position_updated event', () => {
+ let state = livePortfolioReducer(INITIAL_LIVE_PORTFOLIO_STATE, {
+ type: 'positions',
+ offers: [offer],
+ });
+
+ state = livePortfolioReducer(state, {
+ type: 'update',
+ update: {
+ kind: 'position_updated',
+ positionId: 'off_1',
+ fields: { status: 'Repaid', amount_repaid: '10500000' },
+ },
+ });
+
+ expect(state.positions[0].status).toBe('Repaid');
+ expect(state.positions[0].amount_repaid).toBe(10_500_000n);
+ });
+
+ it('never crashes the reducer on malformed wire amounts', () => {
+ let state = livePortfolioReducer(INITIAL_LIVE_PORTFOLIO_STATE, {
+ type: 'positions',
+ offers: [offer],
+ });
+
+ state = livePortfolioReducer(state, {
+ type: 'update',
+ update: {
+ kind: 'position_updated',
+ positionId: 'off_1',
+ fields: { amount_repaid: 'not-a-number' },
+ },
+ });
+
+ expect(state.positions[0].amount_repaid).toBe(0n);
+ });
+
+ it('ignores updates for positions it does not know about', () => {
+ const state = livePortfolioReducer(INITIAL_LIVE_PORTFOLIO_STATE, {
+ type: 'positions',
+ offers: [offer],
+ });
+
+ const next = livePortfolioReducer(state, {
+ type: 'update',
+ update: { kind: 'repayment_received', positionId: 'off_unknown', amountRepaid: 1n, fullyRepaid: false },
+ });
+
+ expect(next).toBe(state);
+ });
+
+ it('records the connection status and transport', () => {
+ const state = livePortfolioReducer(INITIAL_LIVE_PORTFOLIO_STATE, {
+ type: 'connection',
+ connection: 'polling',
+ transport: 'polling',
+ detail: 'relay unreachable',
+ });
+
+ expect(state.connection).toBe('polling');
+ expect(state.transport).toBe('polling');
+ expect(state.connectionDetail).toBe('relay unreachable');
+ });
+});
\ No newline at end of file
diff --git a/invofi/apps/frontend/src/lib/live/reducer.ts b/invofi/apps/frontend/src/lib/live/reducer.ts
new file mode 100644
index 000000000..b13496a93
--- /dev/null
+++ b/invofi/apps/frontend/src/lib/live/reducer.ts
@@ -0,0 +1,169 @@
+// ── Live portfolio reducer (issue #221) ──────────────────────────────────────
+// Pure state transitions for the dashboard. The context provider dispatches
+// here from the engine; components read the resulting `LivePosition`s. Because
+// every action goes through this reducer, tests can drive the full data flow
+// without React.
+
+import type { FinancingOffer } from '@invofi/sdk';
+import { toStroopsBigInt } from '@/lib/utils';
+import { safeStroopsFromWire } from './convert';
+import type { ConnectionStatus, LivePosition, LivePositionUpdate, LiveTransport } from './types';
+import { offerApy, remainingStroops, repaymentProgress, totalDueStroops, yieldEarnedStroops } from './yield';
+import { usdPriceFor } from './prices';
+import { STROOPS_PER_XLM } from '@/lib/constants';
+
+export interface LivePortfolioState {
+ connection: ConnectionStatus;
+ /** Optional human-readable detail for the status pill (e.g. "retry in 2s"). */
+ connectionDetail: string | null;
+ transport: LiveTransport;
+ positions: LivePosition[];
+ loading: boolean;
+ error: string | null;
+ /** Epoch ms of the last data-changing event (for "updated Xs ago" hints). */
+ lastUpdatedAt: number | null;
+}
+
+export type LivePortfolioAction =
+ | { type: 'connection'; connection: ConnectionStatus; transport: LiveTransport; detail?: string | null }
+ | { type: 'positions'; offers: FinancingOffer[] }
+ | { type: 'update'; update: LivePositionUpdate }
+ | { type: 'loading'; loading: boolean }
+ | { type: 'error'; error: string | null }
+ | { type: 'reset' };
+
+export const INITIAL_LIVE_PORTFOLIO_STATE: LivePortfolioState = {
+ connection: 'connecting',
+ connectionDetail: null,
+ transport: 'websocket',
+ positions: [],
+ loading: true,
+ error: null,
+ lastUpdatedAt: null,
+};
+
+/** Compute every derived live field for a single position row. */
+export function deriveLivePosition(
+ offer: FinancingOffer,
+ nowSecs: number = Date.now() / 1000,
+): LivePosition {
+ const price = usdPriceFor(offer.currency);
+ const remaining = remainingStroops(offer);
+ return {
+ ...offer,
+ amount: toStroopsBigInt(offer.amount),
+ amount_repaid: toStroopsBigInt(offer.amount_repaid),
+ totalDue: totalDueStroops(offer),
+ remaining,
+ repaymentProgress: repaymentProgress(offer),
+ apy: offerApy(offer),
+ earnedToDate: yieldEarnedStroops(offer, nowSecs),
+ liveValueUsd: (Number(remaining) / STROOPS_PER_XLM) * price,
+ updatedAt: Date.now(),
+ };
+}
+
+/** Build a live row from a stored row + an update, or undefined if unknown. */
+function applyUpdate(
+ position: LivePosition | undefined,
+ update: LivePositionUpdate,
+): LivePosition | undefined {
+ if (!position) return undefined;
+ const now = update.updatedAt ?? Date.now();
+
+ switch (update.kind) {
+ case 'position_updated': {
+ const merged: FinancingOffer = {
+ ...position,
+ status: update.fields.status ?? position.status,
+ funded_at:
+ update.fields.funded_at !== undefined ? Number(update.fields.funded_at) : position.funded_at,
+ currency: update.fields.currency ?? position.currency,
+ amount: safeStroopsFromWire(update.fields.amount ?? position.amount),
+ amount_repaid: safeStroopsFromWire(update.fields.amount_repaid ?? position.amount_repaid),
+ };
+ const derived = deriveLivePosition(merged, now / 1000);
+ return { ...derived, updatedAt: now };
+ }
+
+ case 'yield_calculated':
+ return {
+ ...position,
+ apy: update.apy,
+ earnedToDate: update.earnedToDate,
+ updatedAt: now,
+ };
+
+ case 'repayment_received': {
+ let amountRepaid: bigint;
+ if (update.remaining !== undefined) {
+ // Cumulative outstanding claim: monotonic, so a replayed or stale
+ // delivery can never double-count a repayment.
+ const derived = totalDueStroops(position) - update.remaining;
+ amountRepaid = derived > position.amount_repaid ? derived : position.amount_repaid;
+ } else {
+ amountRepaid = position.amount_repaid + update.amountRepaid;
+ }
+ const merged = {
+ ...position,
+ amount_repaid: amountRepaid,
+ };
+ const derived = deriveLivePosition(merged, now / 1000);
+ return {
+ ...derived,
+ status: update.fullyRepaid ? 'Repaid' : derived.status,
+ updatedAt: now,
+ };
+ }
+ }
+}
+
+export function livePortfolioReducer(
+ state: LivePortfolioState,
+ action: LivePortfolioAction,
+): LivePortfolioState {
+ switch (action.type) {
+ case 'connection':
+ return {
+ ...state,
+ connection: action.connection,
+ transport: action.transport,
+ connectionDetail: action.detail ?? null,
+ };
+
+ case 'positions':
+ return {
+ ...state,
+ positions: action.offers.map(offer => deriveLivePosition(offer)),
+ loading: false,
+ error: null,
+ lastUpdatedAt: Date.now(),
+ };
+
+ case 'update': {
+ const { update } = action;
+ let changed = false;
+ const positions = state.positions.map(position => {
+ if (position.id !== update.positionId) return position;
+ const next = applyUpdate(position, update);
+ if (!next) return position;
+ changed = true;
+ return next;
+ });
+ if (!changed) return state;
+ return { ...state, positions, lastUpdatedAt: Date.now() };
+ }
+
+ case 'loading':
+ return { ...state, loading: action.loading };
+
+ case 'error':
+ return { ...state, error: action.error };
+
+ case 'reset':
+ return INITIAL_LIVE_PORTFOLIO_STATE;
+
+ default:
+ return state;
+ }
+}
\ No newline at end of file
diff --git a/invofi/apps/frontend/src/lib/live/throttle.test.ts b/invofi/apps/frontend/src/lib/live/throttle.test.ts
new file mode 100644
index 000000000..ae201b21c
--- /dev/null
+++ b/invofi/apps/frontend/src/lib/live/throttle.test.ts
@@ -0,0 +1,91 @@
+import { afterEach, describe, expect, it, vi } from 'vitest';
+import { createPerKeyThrottle } from './throttle';
+
+describe('createPerKeyThrottle', () => {
+ afterEach(() => {
+ vi.useRealTimers();
+ });
+
+ it('delivers the leading edge immediately, then coalesces the window', () => {
+ vi.useFakeTimers();
+ const delivered: Array<[string, number]> = [];
+ const throttle = createPerKeyThrottle(1_000, (key, value) => delivered.push([key, value]));
+
+ throttle('a', 1); // leading edge — delivered now
+ throttle('a', 2);
+ throttle('a', 3);
+ throttle('b', 10); // leading edge — delivered now
+
+ expect(delivered).toEqual([
+ ['a', 1],
+ ['b', 10],
+ ]);
+
+ vi.advanceTimersByTime(1_000);
+
+ expect(delivered).toEqual([
+ ['a', 1],
+ ['b', 10],
+ ['a', 3],
+ ]);
+ });
+
+ it('delivers at most once per interval per key', () => {
+ vi.useFakeTimers();
+ const delivered: Array<[string, number]> = [];
+ const throttle = createPerKeyThrottle(1_000, (key, value) => delivered.push([key, value]));
+
+ throttle('a', 0); // leading edge at t=0
+ for (let i = 1; i < 6; i++) {
+ throttle('a', i);
+ vi.advanceTimersByTime(100);
+ }
+ // Window opened at t=0 closes at t=1000, delivering the coalesced latest.
+ vi.advanceTimersByTime(1_000);
+ expect(delivered).toEqual([
+ ['a', 0],
+ ['a', 5],
+ ]);
+ });
+
+ it('keeps per-key windows independent', () => {
+ vi.useFakeTimers();
+ const delivered: string[] = [];
+ const throttle = createPerKeyThrottle(1_000, key => delivered.push(key));
+
+ throttle('a', '1');
+ vi.advanceTimersByTime(500);
+ throttle('b', '2');
+ vi.advanceTimersByTime(500); // a window closes (nothing pending); b not yet due
+ throttle('a', '3'); // a window is open again → leading edge
+ vi.advanceTimersByTime(1_000); // b window closes (nothing pending)
+
+ expect(delivered).toEqual(['a', 'b', 'a']);
+ });
+
+ it('flush delivers pending values immediately and cancels armed timers', () => {
+ vi.useFakeTimers();
+ const delivered: number[] = [];
+ const throttle = createPerKeyThrottle(1_000, (_key, value) => delivered.push(value));
+
+ throttle('a', 1); // leading edge
+ throttle('a', 2); // pending
+ throttle.flush();
+ vi.advanceTimersByTime(2_000);
+
+ expect(delivered).toEqual([1, 2]);
+ });
+
+ it('stop cancels timers and drops pending values', () => {
+ vi.useFakeTimers();
+ const delivered: number[] = [];
+ const throttle = createPerKeyThrottle(1_000, (_key, value) => delivered.push(value));
+
+ throttle('a', 1); // leading edge already delivered
+ throttle('a', 2); // pending — dropped by stop()
+ throttle.stop();
+ vi.advanceTimersByTime(2_000);
+
+ expect(delivered).toEqual([1]);
+ });
+});
\ No newline at end of file
diff --git a/invofi/apps/frontend/src/lib/live/throttle.ts b/invofi/apps/frontend/src/lib/live/throttle.ts
new file mode 100644
index 000000000..4159e8cc1
--- /dev/null
+++ b/invofi/apps/frontend/src/lib/live/throttle.ts
@@ -0,0 +1,70 @@
+// ── Per-position update throttle (issue #221) ────────────────────────────────
+// Live sources can burst many updates for the same position in a single second
+// (a repayment transaction fires multiple on-chain events, a relay may fan out
+// several messages). The acceptance criteria cap UI churn at one update per
+// position per second. The first update for a key is delivered immediately
+// (leading edge); further updates within the window coalesce, and the *latest*
+// is delivered when the window closes.
+
+export interface PerKeyThrottle {
+ /** Queue `value` for `key`, coalescing with anything already pending. */
+ (key: string, value: T): void;
+ /** Immediately deliver everything still pending. Used on shutdown. */
+ flush: () => void;
+ /** Cancel pending timers and drop queued values. */
+ stop: () => void;
+}
+
+/**
+ * Returns a throttled dispatcher that calls `deliver` at most once per `key`
+ * per `intervalMs`, always with the most recent value.
+ */
+export function createPerKeyThrottle(
+ intervalMs: number,
+ deliver: (key: string, value: T) => void,
+): PerKeyThrottle {
+ const pending = new Map();
+ const timers = new Map>();
+ let stopped = false;
+
+ function dispatch(key: string, value: T): void {
+ if (stopped) return;
+ if (!timers.has(key)) {
+ // Leading edge: deliver now, then open the coalescing window. The cap is
+ // one delivery per interval per key — not a minimum one-second latency.
+ deliver(key, value);
+ timers.set(
+ key,
+ setTimeout(() => {
+ timers.delete(key);
+ if (!pending.has(key)) return;
+ const latest = pending.get(key) as T;
+ pending.delete(key);
+ dispatch(key, latest);
+ }, intervalMs),
+ );
+ return;
+ }
+ pending.set(key, value);
+ }
+
+ dispatch.flush = () => {
+ if (stopped) return;
+ // Cancel armed timers so nothing fires after the flush.
+ for (const timer of timers.values()) clearTimeout(timer);
+ timers.clear();
+ for (const [key, value] of pending) {
+ pending.delete(key);
+ deliver(key, value);
+ }
+ };
+
+ dispatch.stop = () => {
+ stopped = true;
+ for (const timer of timers.values()) clearTimeout(timer);
+ timers.clear();
+ pending.clear();
+ };
+
+ return dispatch;
+}
\ No newline at end of file
diff --git a/invofi/apps/frontend/src/lib/live/transports.test.ts b/invofi/apps/frontend/src/lib/live/transports.test.ts
new file mode 100644
index 000000000..3742d52ec
--- /dev/null
+++ b/invofi/apps/frontend/src/lib/live/transports.test.ts
@@ -0,0 +1,436 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import { createPollingTransport, createWebSocketTransport, decodeEnvelope } from './transports';
+
+// ── SDK mocking ──────────────────────────────────────────────────────────────
+// The polling transport delegates to the SDK's listenToEvents; stub it so we
+// can assert wiring without a real RPC.
+
+const { listenToEventsMock } = vi.hoisted(() => ({
+ listenToEventsMock: vi.fn(),
+}));
+
+vi.mock('@invofi/sdk', () => ({
+ listenToEvents: listenToEventsMock,
+}));
+
+// ── Fake WebSocket for the relay transport tests ────────────────────────────
+
+class FakeWebSocket {
+ static instances: FakeWebSocket[] = [];
+ url: string;
+ onopen: (() => void) | null = null;
+ onclose: (() => void) | null = null;
+ onerror: ((event: unknown) => void) | null = null;
+ onmessage: ((event: { data: string }) => void) | null = null;
+ closed = false;
+
+ constructor(url: string) {
+ this.url = url;
+ FakeWebSocket.instances.push(this);
+ }
+
+ close() {
+ this.closed = true;
+ }
+
+ emitOpen() {
+ this.onopen?.();
+ }
+
+ emitClose() {
+ this.onclose?.();
+ }
+
+ emitMessage(data: string) {
+ this.onmessage?.({ data });
+ }
+}
+
+describe('decodeEnvelope', () => {
+ it('decodes position_updated with a received-time stamp', () => {
+ expect(
+ decodeEnvelope(
+ JSON.stringify({ type: 'position_updated', positionId: 'off_1', fields: { status: 'Financed' } }),
+ ),
+ ).toEqual({
+ kind: 'position_updated',
+ positionId: 'off_1',
+ fields: { status: 'Financed' },
+ updatedAt: expect.any(Number),
+ });
+ });
+
+ it('decodes yield_calculated with string bigint amounts', () => {
+ expect(
+ decodeEnvelope(
+ JSON.stringify({ type: 'yield_calculated', positionId: 'off_1', apy: 12.5, earnedToDate: '250000' }),
+ ),
+ ).toEqual({
+ kind: 'yield_calculated',
+ positionId: 'off_1',
+ apy: 12.5,
+ earnedToDate: 250_000n,
+ updatedAt: expect.any(Number),
+ });
+ });
+
+ it('decodes repayment_received and forwards the cumulative remaining', () => {
+ expect(
+ decodeEnvelope(
+ JSON.stringify({
+ type: 'repayment_received',
+ positionId: 'off_1',
+ amountRepaid: '1000000',
+ remaining: '5000000',
+ progress: 0.2,
+ fullyRepaid: false,
+ }),
+ ),
+ ).toEqual({
+ kind: 'repayment_received',
+ positionId: 'off_1',
+ amountRepaid: 1_000_000n,
+ remaining: 5_000_000n,
+ fullyRepaid: false,
+ updatedAt: expect.any(Number),
+ });
+ });
+
+ it('accepts repayment envelopes that omit progress', () => {
+ const decoded = decodeEnvelope(
+ JSON.stringify({
+ type: 'repayment_received',
+ positionId: 'off_1',
+ amountRepaid: '1000000',
+ fullyRepaid: false,
+ }),
+ );
+ expect(decoded).not.toBeNull();
+ expect(decoded && decoded.kind).toBe('repayment_received');
+ });
+
+ it('rejects repayment envelopes with malformed or unsafe amounts', () => {
+ expect(
+ decodeEnvelope(
+ JSON.stringify({ type: 'repayment_received', positionId: 'off_1', amountRepaid: {}, fullyRepaid: false }),
+ ),
+ ).toBeNull();
+ // Above Number.MAX_SAFE_INTEGER → stroopsFromWire throws → caught → null.
+ expect(
+ decodeEnvelope(
+ JSON.stringify({ type: 'repayment_received', positionId: 'off_1', amountRepaid: 9007199254740993, fullyRepaid: false }),
+ ),
+ ).toBeNull();
+ });
+
+ it('returns null for malformed or unknown envelopes', () => {
+ expect(decodeEnvelope('not json')).toBeNull();
+ expect(decodeEnvelope('{}')).toBeNull();
+ expect(decodeEnvelope(JSON.stringify({ type: 'nope', positionId: 'off_1' }))).toBeNull();
+ });
+});
+
+describe('createWebSocketTransport', () => {
+ beforeEach(() => {
+ FakeWebSocket.instances = [];
+ vi.stubGlobal('WebSocket', FakeWebSocket);
+ vi.useFakeTimers();
+ });
+
+ afterEach(() => {
+ vi.unstubAllGlobals();
+ vi.useRealTimers();
+ });
+
+ it('connects, reports connected, and forwards relay updates', () => {
+ const updates: unknown[] = [];
+ const statuses: string[] = [];
+ const transport = createWebSocketTransport({
+ url: 'wss://relay.invofi.dev',
+ onUpdate: update => updates.push(update),
+ onConnectionChange: status => statuses.push(status),
+ onResync: () => {},
+ onGiveUp: () => {},
+ });
+
+ transport.start();
+ FakeWebSocket.instances[0].emitOpen();
+ FakeWebSocket.instances[0].emitMessage(
+ JSON.stringify({ type: 'repayment_received', positionId: 'off_1', amountRepaid: 5, progress: 0, fullyRepaid: false }),
+ );
+
+ expect(statuses).toEqual(['connecting', 'connected']);
+ expect(updates).toHaveLength(1);
+ });
+
+ it('reconnects with exponential backoff after a drop (≤ 5s first retry)', () => {
+ const statuses: string[] = [];
+ const transport = createWebSocketTransport({
+ url: 'wss://relay.invofi.dev',
+ onUpdate: () => {},
+ onConnectionChange: status => statuses.push(status),
+ onResync: () => {},
+ onGiveUp: () => {},
+ reconnectBaseMs: 1_000,
+ maxReconnectAttempts: 3,
+ });
+
+ transport.start();
+ FakeWebSocket.instances[0].emitOpen(); // connected once
+ FakeWebSocket.instances[0].emitClose(); // drop
+
+ expect(statuses).toEqual(['connecting', 'connected', 'reconnecting']);
+
+ vi.advanceTimersByTime(999);
+ expect(FakeWebSocket.instances).toHaveLength(1); // not yet reconnected
+ vi.advanceTimersByTime(1);
+ expect(FakeWebSocket.instances).toHaveLength(2); // first retry within 1s
+
+ FakeWebSocket.instances[1].emitOpen();
+ FakeWebSocket.instances[1].emitClose();
+ vi.advanceTimersByTime(2_000); // backoff doubled → 2s
+ expect(FakeWebSocket.instances).toHaveLength(3);
+ });
+
+ it('gives up to the polling fallback after repeated failed initial attempts', () => {
+ const giveUp: string[] = [];
+ const transport = createWebSocketTransport({
+ url: 'wss://relay.invofi.dev',
+ onUpdate: () => {},
+ onConnectionChange: () => {},
+ onResync: () => {},
+ onGiveUp: reason => giveUp.push(reason),
+ reconnectBaseMs: 1_000,
+ maxReconnectAttempts: 3,
+ });
+
+ transport.start(); // attempt 1 (instance 0)
+ FakeWebSocket.instances[0].emitClose();
+ vi.advanceTimersByTime(2_000); // → attempt 2 (instance 1)
+ FakeWebSocket.instances[1].emitClose();
+ vi.advanceTimersByTime(4_000); // → attempt 3 (instance 2)
+ FakeWebSocket.instances[2].emitClose();
+
+ expect(giveUp).toHaveLength(1);
+ expect(FakeWebSocket.instances).toHaveLength(3); // no further attempts
+ });
+
+ it('hands off to polling after repeated failures once connected', () => {
+ const giveUp: string[] = [];
+ const statuses: string[] = [];
+ const transport = createWebSocketTransport({
+ url: 'wss://relay.invofi.dev',
+ onUpdate: () => {},
+ onConnectionChange: status => statuses.push(status),
+ onResync: () => {},
+ onGiveUp: reason => giveUp.push(reason),
+ reconnectBaseMs: 1_000,
+ maxReconnectAttempts: 3,
+ maxRelayFailures: 2,
+ });
+
+ transport.start();
+ FakeWebSocket.instances[0].emitOpen(); // connected
+ FakeWebSocket.instances[0].emitClose(); // failure 1 → reconnect at 1s
+ vi.advanceTimersByTime(1_000);
+ expect(FakeWebSocket.instances).toHaveLength(2);
+
+ FakeWebSocket.instances[1].emitOpen();
+ FakeWebSocket.instances[1].emitClose(); // failure 2 ≥ maxRelayFailures → give up
+ expect(giveUp).toEqual(['connection closed']);
+ vi.advanceTimersByTime(10_000);
+ expect(FakeWebSocket.instances).toHaveLength(2); // no more reconnects
+ });
+
+ it('treats a stalled handshake as a failed connect', () => {
+ const statuses: string[] = [];
+ const transport = createWebSocketTransport({
+ url: 'wss://relay.invofi.dev',
+ onUpdate: () => {},
+ onConnectionChange: status => statuses.push(status),
+ onResync: () => {},
+ onGiveUp: () => {},
+ connectTimeoutMs: 1_000,
+ });
+
+ transport.start();
+ expect(FakeWebSocket.instances).toHaveLength(1);
+
+ vi.advanceTimersByTime(1_000); // no onopen ever fires
+ expect(FakeWebSocket.instances[0].closed).toBe(true);
+ expect(statuses).toContain('connecting');
+ // A retry is scheduled after the timed-out attempt.
+ vi.advanceTimersByTime(2_000);
+ expect(FakeWebSocket.instances).toHaveLength(2);
+ });
+
+ it('stop() cancels pending reconnects and detaches handlers', () => {
+ const updates: unknown[] = [];
+ const transport = createWebSocketTransport({
+ url: 'wss://relay.invofi.dev',
+ onUpdate: update => updates.push(update),
+ onConnectionChange: () => {},
+ onResync: () => {},
+ onGiveUp: () => {},
+ reconnectBaseMs: 1_000,
+ maxReconnectAttempts: 3,
+ });
+
+ transport.start();
+ const first = FakeWebSocket.instances[0];
+ first.emitOpen();
+ first.emitClose(); // schedules a reconnect
+ transport.stop();
+
+ vi.advanceTimersByTime(10_000);
+ expect(FakeWebSocket.instances).toHaveLength(1);
+
+ first.emitMessage(
+ JSON.stringify({ type: 'position_updated', positionId: 'off_1', fields: {} }),
+ );
+ expect(updates).toHaveLength(0);
+ });
+});
+
+describe('createPollingTransport', () => {
+ beforeEach(() => {
+ listenToEventsMock.mockClear();
+ });
+
+ afterEach(() => {
+ vi.useRealTimers();
+ });
+
+ it('subscribes to protocol events and reports polling status', () => {
+ const statuses: string[] = [];
+ const transport = createPollingTransport({
+ contractIds: ['registry', 'financing', 'repayment'],
+ rpcUrl: 'https://soroban-testnet.stellar.org',
+ networkPassphrase: 'testnet',
+ onUpdate: () => {},
+ onConnectionChange: status => statuses.push(status),
+ onResync: () => {},
+ });
+
+ transport.start();
+
+ expect(statuses).toEqual(['polling']);
+ expect(listenToEventsMock).toHaveBeenCalledWith(
+ expect.objectContaining({
+ eventTypes: expect.arrayContaining(['inv_rep', 'off_acc']),
+ contractIds: ['registry', 'financing', 'repayment'],
+ rpcUrl: 'https://soroban-testnet.stellar.org',
+ networkPassphrase: 'testnet',
+ pollIntervalMs: 5_000,
+ }),
+ );
+ });
+
+ it('maps inv_rep events to repayment updates without an extra resync', () => {
+ type TestEvent = {
+ type: string;
+ subjectId: string;
+ data: { amount: bigint; fullyRepaid: boolean };
+ };
+ const updates: unknown[] = [];
+ let resyncs = 0;
+ const captured: { onEvent: ((event: TestEvent) => void) | null } = { onEvent: null };
+ listenToEventsMock.mockImplementation((opts: { onEvent: (event: TestEvent) => void }) => {
+ captured.onEvent = opts.onEvent;
+ return () => {};
+ });
+
+ const transport = createPollingTransport({
+ contractIds: ['repayment'],
+ rpcUrl: 'https://soroban-testnet.stellar.org',
+ networkPassphrase: 'testnet',
+ onUpdate: update => updates.push(update),
+ onConnectionChange: () => {},
+ onResync: () => resyncs++,
+ });
+
+ transport.start();
+ captured.onEvent?.({
+ type: 'inv_rep',
+ subjectId: 'off_1',
+ data: { amount: 1_000_000n, fullyRepaid: false },
+ });
+
+ expect(updates).toEqual([
+ {
+ kind: 'repayment_received',
+ positionId: 'off_1',
+ amountRepaid: 1_000_000n,
+ fullyRepaid: false,
+ updatedAt: expect.any(Number),
+ },
+ ]);
+ expect(resyncs).toBe(0);
+ });
+
+ it('coalesces a batch of non-repayment events into a single resync', () => {
+ vi.useFakeTimers();
+ type TestEvent = { type: string; subjectId: string };
+ const captured: { onEvent: ((event: TestEvent) => void) | null } = { onEvent: null };
+ listenToEventsMock.mockImplementation((opts: { onEvent: (event: TestEvent) => void }) => {
+ captured.onEvent = opts.onEvent;
+ return () => {};
+ });
+ let resyncs = 0;
+
+ const transport = createPollingTransport({
+ contractIds: ['registry'],
+ rpcUrl: 'https://soroban-testnet.stellar.org',
+ networkPassphrase: 'testnet',
+ onUpdate: () => {},
+ onConnectionChange: () => {},
+ onResync: () => resyncs++,
+ });
+
+ transport.start();
+ captured.onEvent?.({ type: 'off_acc', subjectId: 'off_1' });
+ captured.onEvent?.({ type: 'off_acc', subjectId: 'off_2' });
+ expect(resyncs).toBe(0);
+
+ vi.advanceTimersByTime(0); // debounce fires once for the whole batch
+ expect(resyncs).toBe(1);
+ });
+
+ it('forwards poll errors to the connection detail', () => {
+ const statuses: string[] = [];
+ const captured: { onError: ((error: Error) => void) | null } = { onError: null };
+ listenToEventsMock.mockImplementation((opts: { onError: (error: Error) => void }) => {
+ captured.onError = opts.onError;
+ return () => {};
+ });
+
+ const transport = createPollingTransport({
+ contractIds: ['repayment'],
+ rpcUrl: 'https://soroban-testnet.stellar.org',
+ networkPassphrase: 'testnet',
+ onUpdate: () => {},
+ onConnectionChange: (status, detail) => statuses.push(status, detail ?? ''),
+ onResync: () => {},
+ });
+
+ transport.start();
+ captured.onError?.(new Error('rpc unreachable'));
+
+ expect(statuses).toContain('polling');
+ expect(statuses).toContain('poll error: rpc unreachable');
+ });
+
+ it('does not call listenToEvents when no contracts are configured', () => {
+ const transport = createPollingTransport({
+ contractIds: [],
+ rpcUrl: 'https://soroban-testnet.stellar.org',
+ networkPassphrase: 'testnet',
+ onUpdate: () => {},
+ onConnectionChange: () => {},
+ onResync: () => {},
+ });
+
+ transport.start();
+ expect(listenToEventsMock).not.toHaveBeenCalled();
+ });
+});
\ No newline at end of file
diff --git a/invofi/apps/frontend/src/lib/live/transports.ts b/invofi/apps/frontend/src/lib/live/transports.ts
new file mode 100644
index 000000000..e8e9b3a9c
--- /dev/null
+++ b/invofi/apps/frontend/src/lib/live/transports.ts
@@ -0,0 +1,362 @@
+// ── Live transports (issue #221) ─────────────────────────────────────────────
+// Two interchangeable sources feed the live dashboard:
+//
+// 1. `createWebSocketTransport` — connects to a relay (e.g. the keeper's
+// SSE/WS endpoint, or any server that mirrors on-chain events) that pushes
+// `position_updated`, `yield_calculated`, and `repayment_received`
+// envelopes. Reconnects with exponential backoff (first retry within ~1s)
+// and gives up to the polling fallback after repeated initial failures.
+//
+// 2. `createPollingTransport` — the graceful-degradation path. It subscribes
+// to the Soroban RPC event stream via the SDK's `listenToEvents` (already
+// the project's documented SSE/WS upgrade path — see `apps/sdk/events.ts`)
+// and turns on-chain events into the same normalized updates. A periodic
+// Supabase resync (owned by the engine) fills any gaps.
+//
+// Both produce `LivePositionUpdate`s so the reducer and UI are transport-agnostic.
+
+import { listenToEvents, type ProtocolEventName } from '@invofi/sdk';
+import type { ConnectionStatus, LivePositionUpdate, WsEnvelope } from './types';
+import { stroopsFromWire } from './convert';
+
+// ── Shared handle ────────────────────────────────────────────────────────────
+
+export interface TransportHandle {
+ start: () => void;
+ stop: () => void;
+}
+
+export interface TransportCallbacks {
+ onUpdate: (update: LivePositionUpdate) => void;
+ onConnectionChange: (status: ConnectionStatus, detail?: string) => void;
+ /** Called when a resync of all positions is warranted. */
+ onResync: () => void;
+}
+
+// ── WebSocket relay transport ────────────────────────────────────────────────
+
+export interface WebSocketTransportOptions extends TransportCallbacks {
+ url: string;
+ /** First reconnect delay after a drop. Default 1000ms (reconnect ≤ 5s). */
+ reconnectBaseMs?: number;
+ /** Ceiling for the exponential backoff. Default 30s. */
+ maxBackoffMs?: number;
+ /** How many failed initial connect attempts before giving up to polling. */
+ maxReconnectAttempts?: number;
+ /** Repeated failures after a successful connect before giving up. Default 8. */
+ maxRelayFailures?: number;
+ /** How long to wait for the handshake before treating it as failed. Default 10s. */
+ connectTimeoutMs?: number;
+ /** Called once when the relay is permanently unavailable. */
+ onGiveUp: (reason: string) => void;
+}
+
+/** Map a relay envelope to a normalized `LivePositionUpdate`, or null. */
+export function decodeEnvelope(raw: string): LivePositionUpdate | null {
+ let parsed: WsEnvelope;
+ try {
+ parsed = JSON.parse(raw) as WsEnvelope;
+ } catch {
+ return null;
+ }
+ if (!parsed || typeof parsed !== 'object' || typeof parsed.type !== 'string') return null;
+
+ // Both transports stamp the same received-time shape.
+ const updatedAt = Date.now();
+ try {
+ switch (parsed.type) {
+ case 'position_updated':
+ if (typeof parsed.positionId !== 'string') return null;
+ return {
+ kind: 'position_updated',
+ positionId: parsed.positionId,
+ fields: parsed.fields ?? {},
+ updatedAt,
+ };
+ case 'yield_calculated':
+ if (typeof parsed.positionId !== 'string' || typeof parsed.apy !== 'number') return null;
+ return {
+ kind: 'yield_calculated',
+ positionId: parsed.positionId,
+ apy: parsed.apy,
+ earnedToDate: stroopsFromWire(parsed.earnedToDate),
+ updatedAt,
+ };
+ case 'repayment_received':
+ if (typeof parsed.positionId !== 'string') return null;
+ // The normalized update uses amountRepaid, not progress — accept
+ // updates that omit progress and reject malformed amounts up front.
+ if (
+ typeof parsed.amountRepaid !== 'string' &&
+ typeof parsed.amountRepaid !== 'number' &&
+ typeof parsed.amountRepaid !== 'bigint'
+ ) {
+ return null;
+ }
+ return {
+ kind: 'repayment_received',
+ positionId: parsed.positionId,
+ amountRepaid: stroopsFromWire(parsed.amountRepaid),
+ remaining:
+ parsed.remaining === undefined ? undefined : stroopsFromWire(parsed.remaining),
+ fullyRepaid: parsed.fullyRepaid === true,
+ updatedAt,
+ };
+ default:
+ return null;
+ }
+ } catch {
+ // A malformed amount must never crash the message handler.
+ return null;
+ }
+}
+
+/**
+ * WebSocket relay client with exponential-backoff reconnection.
+ *
+ * Reconnect schedule (base 1s, doubling, capped): 1s → 2s → 4s → … so a
+ * dropped live connection is restored within ~1–5 seconds (acceptance
+ * criterion). If the relay never connects within `maxReconnectAttempts`
+ * attempts, `onGiveUp` fires so the caller can degrade to polling.
+ */
+export function createWebSocketTransport(
+ options: WebSocketTransportOptions,
+): TransportHandle {
+ const {
+ url,
+ onUpdate,
+ onConnectionChange,
+ onResync,
+ onGiveUp,
+ reconnectBaseMs = 1_000,
+ maxBackoffMs = 30_000,
+ maxReconnectAttempts = 3,
+ maxRelayFailures = 8,
+ connectTimeoutMs = 10_000,
+ } = options;
+
+ let socket: WebSocket | null = null;
+ let reconnectTimer: ReturnType | null = null;
+ let connectTimer: ReturnType | null = null;
+ let stopped = false;
+ let everConnected = false;
+ let consecutiveFailures = 0;
+
+ function backoffFor(attempt: number): number {
+ return Math.min(maxBackoffMs, reconnectBaseMs * 2 ** Math.min(attempt, 6));
+ }
+
+ function clearConnectTimer(): void {
+ if (connectTimer) {
+ clearTimeout(connectTimer);
+ connectTimer = null;
+ }
+ }
+
+ function connect(): void {
+ if (stopped) return;
+ onConnectionChange(everConnected ? 'reconnecting' : 'connecting');
+
+ let ws: WebSocket;
+ try {
+ ws = new WebSocket(url);
+ } catch {
+ handleFailure('invalid WebSocket URL');
+ return;
+ }
+ socket = ws;
+
+ // A stalled handshake fires neither onopen nor onclose — bound it.
+ connectTimer = setTimeout(() => {
+ if (stopped || socket !== ws) return;
+ socket = null;
+ try {
+ ws.close();
+ } catch {
+ // already closed
+ }
+ handleFailure('connect timed out');
+ }, connectTimeoutMs);
+
+ ws.onopen = () => {
+ if (stopped || socket !== ws) return;
+ clearConnectTimer();
+ everConnected = true;
+ // Note: the failure counter is NOT reset here. Once live, a drop counts
+ // against `maxRelayFailures` so a relay that keeps failing hands off to
+ // polling instead of reconnecting forever.
+ onConnectionChange('connected');
+ // The relay may have missed events while we were offline — resync.
+ onResync();
+ };
+
+ ws.onmessage = (event: MessageEvent) => {
+ if (stopped || socket !== ws) return;
+ const update = decodeEnvelope(String(event.data));
+ if (update) onUpdate(update);
+ };
+
+ ws.onerror = () => {
+ // onclose follows — the close handler owns retry logic.
+ };
+
+ ws.onclose = () => {
+ if (stopped || socket !== ws) return;
+ clearConnectTimer();
+ socket = null;
+ handleFailure('connection closed');
+ };
+ }
+
+ function handleFailure(reason: string): void {
+ if (stopped) return;
+
+ if (everConnected) {
+ // We were live and dropped — reconnect with backoff, but only so far: a
+ // relay that stays down must hand off to the polling fallback.
+ consecutiveFailures += 1;
+ if (consecutiveFailures >= maxRelayFailures) {
+ onGiveUp(reason);
+ return;
+ }
+ const backoff = backoffFor(consecutiveFailures - 1);
+ onConnectionChange('reconnecting', `retry in ${Math.round(backoff / 1000)}s`);
+ reconnectTimer = setTimeout(connect, backoff);
+ return;
+ }
+
+ // Never connected yet — a few attempts, then hand off to polling.
+ consecutiveFailures += 1;
+ if (consecutiveFailures >= maxReconnectAttempts) {
+ onGiveUp(reason);
+ return;
+ }
+ onConnectionChange('connecting', `attempt ${consecutiveFailures + 1}`);
+ reconnectTimer = setTimeout(connect, backoffFor(consecutiveFailures));
+ }
+
+ function stop(): void {
+ stopped = true;
+ clearConnectTimer();
+ if (reconnectTimer) {
+ clearTimeout(reconnectTimer);
+ reconnectTimer = null;
+ }
+ if (socket) {
+ socket.onopen = socket.onclose = socket.onmessage = socket.onerror = null;
+ try {
+ socket.close();
+ } catch {
+ // already closed
+ }
+ socket = null;
+ }
+ }
+
+ return { start: connect, stop };
+}
+
+// ── Soroban event-stream polling transport (fallback) ───────────────────────
+
+const POLLING_EVENT_TYPES: ProtocolEventName[] = [
+ 'off_acc',
+ 'off_rej',
+ 'off_wdr',
+ 'off_def',
+ 'inv_rep',
+ 'pos_mint',
+];
+
+export interface PollingTransportOptions extends TransportCallbacks {
+ /** Contract IDs to watch. Empty → rely on the engine's Supabase resync. */
+ contractIds: string[];
+ rpcUrl: string;
+ networkPassphrase: string;
+ /** How often to poll the RPC. Default 5000ms (one Stellar ledger ≈ 5s). */
+ pollIntervalMs?: number;
+}
+
+/**
+ * Graceful-degradation transport: watches the Soroban RPC event stream for the
+ * user's protocol events and maps them to the same normalized updates.
+ *
+ * `inv_rep` maps directly to a `repayment_received` update for instant feedback;
+ * every other mutation triggers a resync because status transitions need the
+ * full row. A periodic Supabase resync (in the engine) is the safety net.
+ */
+export function createPollingTransport(
+ options: PollingTransportOptions,
+): TransportHandle {
+ const {
+ contractIds,
+ rpcUrl,
+ networkPassphrase,
+ onUpdate,
+ onConnectionChange,
+ onResync,
+ pollIntervalMs = 5_000,
+ } = options;
+
+ let stopListening: (() => void) | null = null;
+ let stopped = false;
+ let resyncPending: ReturnType | null = null;
+
+ // One poll cycle can deliver a batch of events — coalesce them into a single
+ // resync request (the engine's in-flight guard also dedupes concurrently).
+ function requestResync(): void {
+ if (stopped) return;
+ if (resyncPending) return;
+ resyncPending = setTimeout(() => {
+ resyncPending = null;
+ onResync();
+ }, 0);
+ }
+
+ function start(): void {
+ if (stopped) return;
+ onConnectionChange('polling');
+ if (!contractIds || contractIds.length === 0) return; // resync-only fallback
+ stopListening = listenToEvents({
+ rpcUrl,
+ networkPassphrase,
+ contractIds,
+ eventTypes: POLLING_EVENT_TYPES,
+ pollIntervalMs,
+ onEvent(event) {
+ if (stopped) return;
+ if (event.type === 'inv_rep') {
+ onUpdate({
+ kind: 'repayment_received',
+ positionId: event.subjectId,
+ amountRepaid: event.data.amount,
+ fullyRepaid: event.data.fullyRepaid,
+ updatedAt: Date.now(),
+ });
+ // Instant feedback for repayments; other mutations need a resync.
+ return;
+ }
+ requestResync();
+ },
+ onError(error) {
+ // The SDK retries with back-off internally; surface the failure so the
+ // UI's connection detail shows why live data may be stale.
+ if (stopped) return;
+ onConnectionChange('polling', `poll error: ${error.message}`);
+ },
+ });
+ }
+
+ function stop(): void {
+ stopped = true;
+ if (resyncPending) {
+ clearTimeout(resyncPending);
+ resyncPending = null;
+ }
+ if (stopListening) {
+ stopListening();
+ stopListening = null;
+ }
+ }
+
+ return { start, stop };
+}
\ No newline at end of file
diff --git a/invofi/apps/frontend/src/lib/live/types.ts b/invofi/apps/frontend/src/lib/live/types.ts
new file mode 100644
index 000000000..86c4c992f
--- /dev/null
+++ b/invofi/apps/frontend/src/lib/live/types.ts
@@ -0,0 +1,134 @@
+// ── Live portfolio types (issue #221) ───────────────────────────────────────
+// Shared contracts for the WebSocket + polling live dashboard. These are
+// framework-agnostic so the transport, engine, and reducer layers stay
+// unit-testable without React.
+
+import type { Currency, FinancingOffer } from '@invofi/sdk';
+
+/** Which live transport is currently backing the dashboard. */
+export type LiveTransport = 'websocket' | 'polling';
+
+/**
+ * Connection lifecycle of the live stream.
+ *
+ * | Value | Meaning |
+ * |---------------|---------------------------------------------------------------|
+ * | `connecting` | Establishing the initial WebSocket connection |
+ * | `connected` | Streaming live updates over WebSocket |
+ * | `reconnecting`| WebSocket dropped — retrying with exponential backoff |
+ * | `polling` | WebSocket unavailable — Soroban event stream + Supabase poll |
+ * | `offline` | No live source at all (no wallet, no data) |
+ */
+export type ConnectionStatus =
+ | 'connecting'
+ | 'connected'
+ | 'reconnecting'
+ | 'polling'
+ | 'offline';
+
+/**
+ * A portfolio position enriched with live-computed values. Everything the
+ * dashboard renders derives from this single shape, so a full resync and an
+ * incremental streamed update produce identical rows.
+ */
+export interface LivePosition extends FinancingOffer {
+ /** Annualized yield as a percentage (agreed rate scaled to a year). */
+ apy: number;
+ /** Yield accrued so far, in stroops (principal × rate × elapsed/duration). */
+ earnedToDate: bigint;
+ /** Principal + total agreed yield, in stroops. */
+ totalDue: bigint;
+ /** Outstanding claim in stroops (totalDue − amount_repaid, floor 0). */
+ remaining: bigint;
+ /** Repayment progress as a ratio 0..1 (streaming progress bar). */
+ repaymentProgress: number;
+ /** Current USD value of the outstanding claim. */
+ liveValueUsd: number;
+ /** Epoch ms of the last update that touched this position. */
+ updatedAt: number;
+}
+
+/** Partial offer fields delivered by a `position_updated` event. */
+export type PositionUpdatedFields = Partial<
+ Pick
+> & {
+ /** Amounts may arrive as strings over the wire (bigints don't survive JSON). */
+ amount?: bigint | string;
+ amount_repaid?: bigint | string;
+};
+
+/**
+ * Normalized live update delivered to the reducer. Every transport (WebSocket
+ * relay or Soroban event polling) produces these, so the UI code never cares
+ * where an update came from.
+ */
+export type LivePositionUpdate =
+ | {
+ kind: 'position_updated';
+ positionId: string;
+ fields: PositionUpdatedFields;
+ updatedAt?: number;
+ }
+ | {
+ kind: 'yield_calculated';
+ positionId: string;
+ apy: number;
+ earnedToDate: bigint;
+ updatedAt?: number;
+ }
+ | {
+ kind: 'repayment_received';
+ positionId: string;
+ /** Incremental amount of this repayment, in stroops. */
+ amountRepaid: bigint;
+ /**
+ * Cumulative outstanding claim (totalDue − repaid) when the relay
+ * provides it. When present the reducer derives `amount_repaid` from this
+ * monotonic value so replayed deliveries can't double-count.
+ */
+ remaining?: bigint;
+ fullyRepaid: boolean;
+ updatedAt?: number;
+ };
+
+/**
+ * Wire protocol for a WebSocket relay (see `createWebSocketTransport`).
+ * The relay forwards on-chain events as one of these envelopes:
+ *
+ * ```jsonc
+ * { "type": "position_updated", "positionId": "off_…", "fields": { "status": "Financed", "funded_at": 1713… } }
+ * { "type": "yield_calculated", "positionId": "off_…", "apy": 12.5, "earnedToDate": "1234567" }
+ * { "type": "repayment_received", "positionId": "off_…", "amountRepaid": "1000000", "remaining": "5000000",
+ * "progress": 0.2, "fullyRepaid": false }
+ * ```
+ *
+ * Amounts are stroops and may arrive as JSON numbers or strings (bigints do
+ * not survive JSON) — the transport normalizes them with `toStroopsBigInt`.
+ */
+export type WsEnvelope =
+ | { type: 'position_updated'; positionId: string; fields: PositionUpdatedFields }
+ | {
+ type: 'yield_calculated';
+ positionId: string;
+ apy: number;
+ earnedToDate: bigint | number | string;
+ }
+ | {
+ type: 'repayment_received';
+ positionId: string;
+ amountRepaid: bigint | number | string;
+ remaining: bigint | number | string;
+ progress: number;
+ fullyRepaid: boolean;
+ };
+
+/** Value object describing the current connection for the status pill. */
+export interface ConnectionInfo {
+ status: ConnectionStatus;
+ transport: LiveTransport;
+ /** Human-readable detail (e.g. "retry in 2s" or "relay unavailable"). */
+ detail?: string;
+}
+
+/** Lookup keyed by currency for USD pricing. */
+export type PriceFor = (currency: Currency) => number;
\ No newline at end of file
diff --git a/invofi/apps/frontend/src/lib/live/yield.test.ts b/invofi/apps/frontend/src/lib/live/yield.test.ts
new file mode 100644
index 000000000..e8eb19e0f
--- /dev/null
+++ b/invofi/apps/frontend/src/lib/live/yield.test.ts
@@ -0,0 +1,67 @@
+import { describe, expect, it } from 'vitest';
+import {
+ offerApy,
+ remainingStroops,
+ repaymentProgress,
+ totalDueStroops,
+ totalYieldStroops,
+ yieldEarnedStroops,
+} from './yield';
+
+const DAY = 86_400;
+const offer = {
+ amount: 1_000_000_000n, // 100 XLM
+ amount_repaid: 0n,
+ interest_rate: 500, // 5.00%
+ duration: 30 * DAY,
+ funded_at: 1_000_000,
+};
+
+describe('yield / repayment math', () => {
+ it('computes simple-interest yield and total due', () => {
+ expect(totalYieldStroops(offer)).toBe(50_000_000n); // 5% of principal
+ expect(totalDueStroops(offer)).toBe(1_050_000_000n);
+ });
+
+ it('annualizes the agreed rate into an APY', () => {
+ // 5% over 30 days → ~60.83% over a 365-day year.
+ expect(offerApy(offer)).toBeCloseTo(60.83, 1);
+ expect(offerApy({ interest_rate: 500, duration: 365 * DAY })).toBeCloseTo(5, 5);
+ expect(offerApy({ interest_rate: 500, duration: 0 })).toBe(0);
+ });
+
+ it('accrues yield linearly from funded_at, capped at the total', () => {
+ expect(yieldEarnedStroops(offer, offer.funded_at)).toBe(0n);
+ expect(yieldEarnedStroops(offer, offer.funded_at + 15 * DAY)).toBe(25_000_000n);
+ expect(yieldEarnedStroops(offer, offer.funded_at + 60 * DAY)).toBe(50_000_000n); // capped
+ expect(yieldEarnedStroops({ ...offer, funded_at: 0 })).toBe(0n);
+ expect(yieldEarnedStroops({ ...offer, duration: 0 })).toBe(0n);
+ });
+
+ it('keeps accrual arithmetic in bigint for amounts above Number.MAX_SAFE_INTEGER', () => {
+ // 3_000_000_000_000_000n stroops (~300M XLM) — Number() would round it.
+ const bigOffer = {
+ amount: 3_000_000_000_000_000n,
+ amount_repaid: 0n,
+ interest_rate: 500,
+ duration: 30 * DAY,
+ funded_at: 1_000_000,
+ };
+ // Halfway: exact bigint division, no float rounding.
+ expect(yieldEarnedStroops(bigOffer, bigOffer.funded_at + 15 * DAY)).toBe(75_000_000_000_000n);
+ expect(yieldEarnedStroops(bigOffer, bigOffer.funded_at + 30 * DAY)).toBe(150_000_000_000_000n);
+ });
+
+ it('tracks remaining and repayment progress', () => {
+ expect(remainingStroops(offer)).toBe(1_050_000_000n);
+ expect(repaymentProgress(offer)).toBe(0);
+
+ const halfRepaid = { ...offer, amount_repaid: 525_000_000n };
+ expect(remainingStroops(halfRepaid)).toBe(525_000_000n);
+ expect(repaymentProgress(halfRepaid)).toBeCloseTo(0.5, 5);
+
+ const overpaid = { ...offer, amount_repaid: 2_000_000_000n };
+ expect(remainingStroops(overpaid)).toBe(0n);
+ expect(repaymentProgress(overpaid)).toBe(1);
+ });
+});
\ No newline at end of file
diff --git a/invofi/apps/frontend/src/lib/live/yield.ts b/invofi/apps/frontend/src/lib/live/yield.ts
new file mode 100644
index 000000000..27a62a324
--- /dev/null
+++ b/invofi/apps/frontend/src/lib/live/yield.ts
@@ -0,0 +1,76 @@
+// ── Yield / repayment math (issue #221) ──────────────────────────────────────
+// Pure helpers shared by the reducer and the live engine. The contract uses
+// simple interest: yield = principal × interest_rate / 10000, accreting
+// linearly from `funded_at` across `duration` seconds. All amounts are stroops.
+
+import type { FinancingOffer } from '@invofi/sdk';
+
+export const SECONDS_PER_YEAR = 365 * 86_400;
+
+/** Total agreed yield in stroops: principal × rate (simple interest). */
+export function totalYieldStroops(
+ offer: Pick,
+): bigint {
+ return (offer.amount * BigInt(offer.interest_rate)) / 10_000n;
+}
+
+/** Amount the borrower owes in total: principal + agreed yield. */
+export function totalDueStroops(
+ offer: Pick,
+): bigint {
+ return offer.amount + totalYieldStroops(offer);
+}
+
+/** Outstanding claim in stroops (floored at 0). */
+export function remainingStroops(
+ offer: Pick,
+): bigint {
+ const due = totalDueStroops(offer);
+ const repaid = offer.amount_repaid;
+ return due > repaid ? due - repaid : 0n;
+}
+
+/** Repayment progress as a ratio 0..1 for the streaming progress bar. */
+export function repaymentProgress(
+ offer: Pick,
+): number {
+ const due = totalDueStroops(offer);
+ if (due <= 0n) return 0;
+ const ratio = Number(offer.amount_repaid) / Number(due);
+ return Math.min(1, Math.max(0, ratio));
+}
+
+/**
+ * Annualized yield as a percentage. The agreed rate covers `duration` seconds,
+ * so scaling it to a 365-day year gives a comparable APY across positions.
+ */
+export function offerApy(
+ offer: Pick,
+): number {
+ if (offer.duration <= 0 || offer.interest_rate <= 0) return 0;
+ return (offer.interest_rate / 10_000) * (SECONDS_PER_YEAR / offer.duration) * 100;
+}
+
+/**
+ * Yield accrued so far, in stroops. Linear from `funded_at`, capped at the
+ * total agreed yield. Returns 0 for offers that were never funded.
+ */
+export function yieldEarnedStroops(
+ offer: Pick,
+ nowSecs: number = Date.now() / 1000,
+): bigint {
+ if (offer.funded_at <= 0 || offer.duration <= 0) return 0n;
+ const total = totalYieldStroops(offer);
+ if (total <= 0n) return 0n;
+ const elapsed = Math.max(0, nowSecs - offer.funded_at);
+ if (elapsed >= offer.duration) return total;
+ // Stay in bigint — Number(total) loses precision for valid large on-chain
+ // amounts, and the floor must apply to the integer ratio, not a float.
+ const elapsedSecs = BigInt(Math.floor(elapsed));
+ return (total * elapsedSecs) / BigInt(offer.duration);
+}
+
+/** Whether an offer is actively deploying capital (its yield accrues). */
+export function isActiveOffer(offer: Pick): boolean {
+ return offer.status === 'Accepted' || offer.status === 'Financed';
+}
\ No newline at end of file
diff --git a/invofi/apps/frontend/vitest.config.ts b/invofi/apps/frontend/vitest.config.ts
index 3b3217ecf..a416a1d0b 100644
--- a/invofi/apps/frontend/vitest.config.ts
+++ b/invofi/apps/frontend/vitest.config.ts
@@ -18,6 +18,13 @@ export default defineConfig({
'src/lib/utils.ts',
'src/lib/csv.ts',
'src/lib/constants.ts',
+ 'src/lib/live/throttle.ts',
+ 'src/lib/live/yield.ts',
+ 'src/lib/live/prices.ts',
+ 'src/lib/live/reducer.ts',
+ 'src/lib/live/transports.ts',
+ 'src/lib/live/engine.ts',
+ 'src/lib/live/convert.ts',
'src/hooks/useDebounce.ts',
'src/hooks/useLocalStorage.ts',
'src/hooks/useMediaQuery.ts',