diff --git a/backend/src/index.ts b/backend/src/index.ts index 4ff0cfdc..cef475d9 100644 --- a/backend/src/index.ts +++ b/backend/src/index.ts @@ -35,7 +35,6 @@ import { import { adminAuth } from "./middleware/adminAuth"; import { deleteStreamById, reconcileStream } from "./services/streamStore"; import { getCache } from "./services/cache"; -import { getStreamStats } from "./services/stats"; import { getStreamMetrics } from "./services/streamMetrics"; import { startReconciliationJob } from "./services/reconciliationJob"; @@ -72,7 +71,6 @@ import { StreamStatus, syncStreams, updateStreamStartAt, - getOnChainStreamCount, } from "./services/streamStore"; import { @@ -97,7 +95,6 @@ import { validateEnv } from "./config/validateEnv"; import { getMetricsHistory } from "./services/metricsHistory"; import { register } from "./services/metrics"; import { initCache } from "./services/cache"; -import { getGlobalStats } from "./services/stats"; import { logger } from "./logger"; const STREAM_STATUSES: StreamStatus[] = [ @@ -122,6 +119,156 @@ const ALLOWED_ASSETS = (process.env.ALLOWED_ASSETS || "USDC,XLM") const SORT_FIELDS = ["totalAmount", "startAt", "createdAt", "durationSeconds"] as const; const SORT_ORDERS = ["asc", "desc"] as const; +const PLATFORM_STATS_CACHE_KEY = "analytics:platform-stats"; +const PLATFORM_STATS_CACHE_TTL_SECONDS = 60; +const LEADERBOARD_CACHE_TTL_SECONDS = 5 * 60; + +type LeaderboardType = + | "top_senders" + | "top_recipients" + | "largest_streams"; + +interface PlatformStats { + total_streams: number; + active_streams: number; + completed_streams: number; + canceled_streams: number; + total_vested_by_asset: { + USDC: number; + XLM: number; + }; + unique_senders: number; + unique_recipients: number; +} + +const leaderboardQuerySchema = z.object({ + type: z.enum(["top_senders", "top_recipients", "largest_streams"]), + limit: z.coerce + .number() + .int("limit must be an integer") + .min(1, "limit must be greater than or equal to 1") + .max(50, "limit must be less than or equal to 50") + .default(10), +}); + +function roundAnalyticsAmount(value: number): number { + return Number(value.toFixed(6)); +} + +function buildPlatformStats(): PlatformStats { + const now = nowInSeconds(); + const streams = listStreams(true); + const uniqueSenders = new Set(); + const uniqueRecipients = new Set(); + + const stats: PlatformStats = { + total_streams: streams.length, + active_streams: 0, + completed_streams: 0, + canceled_streams: 0, + total_vested_by_asset: { + USDC: 0, + XLM: 0, + }, + unique_senders: 0, + unique_recipients: 0, + }; + + for (const stream of streams) { + uniqueSenders.add(stream.sender); + uniqueRecipients.add(stream.recipient); + + const progress = calculateProgress(stream, now); + if (progress.status === "active") { + stats.active_streams += 1; + } else if (progress.status === "completed") { + stats.completed_streams += 1; + } else if (progress.status === "canceled") { + stats.canceled_streams += 1; + } + + const assetCode = stream.assetCode.toUpperCase(); + if (assetCode === "USDC" || assetCode === "XLM") { + const vestedAt = stream.canceledAt ?? now; + const vestedAmount = calculateProgress(stream, vestedAt).vestedAmount; + stats.total_vested_by_asset[assetCode] += vestedAmount; + } + } + + stats.total_vested_by_asset.USDC = roundAnalyticsAmount( + stats.total_vested_by_asset.USDC, + ); + stats.total_vested_by_asset.XLM = roundAnalyticsAmount( + stats.total_vested_by_asset.XLM, + ); + stats.unique_senders = uniqueSenders.size; + stats.unique_recipients = uniqueRecipients.size; + + return stats; +} + +function buildLeaderboard(type: LeaderboardType, limit: number) { + const streams = listStreams(true); + + if (type === "top_senders") { + const totals = new Map(); + for (const stream of streams) { + totals.set( + stream.sender, + (totals.get(stream.sender) ?? 0) + stream.totalAmount, + ); + } + + return Array.from(totals.entries()) + .map(([sender, totalAmount]) => ({ + sender, + totalAmount: roundAnalyticsAmount(totalAmount), + })) + .sort( + (a, b) => + b.totalAmount - a.totalAmount || a.sender.localeCompare(b.sender), + ) + .slice(0, limit); + } + + if (type === "top_recipients") { + const totals = new Map(); + for (const stream of streams) { + totals.set( + stream.recipient, + (totals.get(stream.recipient) ?? 0) + stream.totalAmount, + ); + } + + return Array.from(totals.entries()) + .map(([recipient, totalAmount]) => ({ + recipient, + totalAmount: roundAnalyticsAmount(totalAmount), + })) + .sort( + (a, b) => + b.totalAmount - a.totalAmount || + a.recipient.localeCompare(b.recipient), + ) + .slice(0, limit); + } + + return [...streams] + .sort( + (a, b) => + b.totalAmount - a.totalAmount || + a.id.localeCompare(b.id, undefined, { numeric: true }), + ) + .slice(0, limit) + .map((stream) => ({ + id: stream.id, + sender: stream.sender, + recipient: stream.recipient, + assetCode: stream.assetCode, + totalAmount: stream.totalAmount, + })); +} + const listStreamsQuerySchema = z.object({ status: z .string() @@ -386,18 +533,96 @@ app.get("/api/health", (_req: Request, res: Response) => { }); }); -app.get("/api/stats", async (_req: Request, res: Response) => { +app.get("/api/stats", async (req: Request, res: Response) => { + res.set("Cache-Control", `public, max-age=${PLATFORM_STATS_CACHE_TTL_SECONDS}`); + try { - const stats = getGlobalStats(); - const onChainStreamCount = await getOnChainStreamCount(); - res.set("Cache-Control", "max-age=30"); - res.json({ data: { ...stats, onChainStreamCount, localStreamCount: stats.total } }); + try { + const cache = getCache(); + const cached = await cache.get(PLATFORM_STATS_CACHE_KEY); + if (cached) { + res.json({ data: cached }); + return; + } + } catch { + // Cache availability must never make this public endpoint fail. + } + + const stats = buildPlatformStats(); + + try { + const cache = getCache(); + await cache.set( + PLATFORM_STATS_CACHE_KEY, + stats, + PLATFORM_STATS_CACHE_TTL_SECONDS, + ); + } catch { + // Cache availability must never make this public endpoint fail. + } + + res.json({ data: stats }); } catch (error) { - logger.error({ err: error }, "Failed to get stats"); - sendApiError(_req, res, 500, "Failed to compute stats.", { code: "INTERNAL_ERROR" }); + logger.error({ err: error }, "Failed to get platform stats"); + sendApiError(req, res, 500, "Failed to compute stats.", { + code: "INTERNAL_ERROR", + }); } }); +app.get( + "/api/leaderboard", + readLimiter, + async (req: Request, res: Response) => { + const parsedQuery = leaderboardQuerySchema.safeParse(req.query); + if (!parsedQuery.success) { + sendValidationError(req, res, parsedQuery.error.issues); + return; + } + + const { type, limit } = parsedQuery.data; + const cacheKey = `analytics:leaderboard:${type}:${limit}`; + + res.set( + "Cache-Control", + `public, max-age=${LEADERBOARD_CACHE_TTL_SECONDS}`, + ); + + try { + try { + const cache = getCache(); + const cached = await cache.get>( + cacheKey, + ); + if (cached) { + res.set("X-Cache", "HIT"); + res.json({ data: cached, type, limit }); + return; + } + } catch { + // Fall through to the database-backed calculation. + } + + const data = buildLeaderboard(type, limit); + + try { + const cache = getCache(); + await cache.set(cacheKey, data, LEADERBOARD_CACHE_TTL_SECONDS); + } catch { + // Cache availability must never make the endpoint fail. + } + + res.set("X-Cache", "MISS"); + res.json({ data, type, limit }); + } catch (error) { + logger.error({ err: error, type, limit }, "Failed to get leaderboard"); + sendApiError(req, res, 500, "Failed to compute leaderboard.", { + code: "INTERNAL_ERROR", + }); + } + }, +); + const METRICS_AUTH = process.env.METRICS_AUTH?.trim() || null; // format: "user:password" app.get("/metrics", async (_req: Request, res: Response) => { @@ -1972,36 +2197,72 @@ if (require.main === module) { }); } -app.delete("/api/streams/:id", adminAuth, (req: Request, res: Response) => { - const parsedId = parseStreamId(req.params.id); - if (!parsedId.ok) { - sendValidationError(req, res, parsedId.issues); - return; - } +app.delete( + "/api/streams/:id", + mutationLimiter, + authMiddleware, + async (req: Request, res: Response) => { + const parsedId = parseStreamId(req.params.id); + if (!parsedId.ok) { + sendValidationError(req, res, parsedId.issues); + return; + } - try { - const deleted = deleteStreamById(parsedId.value); + const stream = getStream(parsedId.value); + if (!stream) { + sendApiError(req, res, 404, "Stream not found.", { code: "NOT_FOUND" }); + return; + } - if (!deleted) { - sendApiError(req, res, 404, "Stream not found or already archived.", { code: "NOT_FOUND" }); + const user = (req as any).user; + if (stream.sender !== user.accountId) { + sendApiError(req, res, 403, "Only the sender can archive this stream.", { + code: "FORBIDDEN", + }); return; } - res.status(204).send(); - } catch (error: any) { - logger.error({ err: error, streamId: parsedId.value }, "failed to delete stream"); - const normalizedError = normalizeUnknownApiError( - error, - "Failed to delete stream.", - ); - sendApiError( - req, - res, - normalizedError.statusCode, - normalizedError.message, - { - code: normalizedError.code ?? "INTERNAL_ERROR", - }, - ); - } -}); + const status = calculateProgress(stream).status; + if (status !== "completed" && status !== "canceled") { + sendApiError( + req, + res, + 400, + "Only completed or canceled streams can be archived.", + { code: "INVALID_STREAM_STATUS" }, + ); + return; + } + + try { + const archived = await deleteStreamById(parsedId.value); + + if (!archived) { + sendApiError(req, res, 404, "Stream not found or already archived.", { + code: "NOT_FOUND", + }); + return; + } + + res.status(204).send(); + } catch (error: any) { + logger.error( + { err: error, streamId: parsedId.value }, + "failed to archive stream", + ); + const normalizedError = normalizeUnknownApiError( + error, + "Failed to archive stream.", + ); + sendApiError( + req, + res, + normalizedError.statusCode, + normalizedError.message, + { + code: normalizedError.code ?? "INTERNAL_ERROR", + }, + ); + } + }, +); diff --git a/backend/src/services/streamStore.ts b/backend/src/services/streamStore.ts index 71eb6d05..66bb2173 100644 --- a/backend/src/services/streamStore.ts +++ b/backend/src/services/streamStore.ts @@ -1313,7 +1313,7 @@ export function markStreamComplete(id: string, at: number = nowInSeconds()): Str return stream; } -export function deleteStreamById(id: string): boolean { +export async function deleteStreamById(id: string): Promise { const db = getDb(); const stream = db @@ -1327,5 +1327,8 @@ export function deleteStreamById(id: string): boolean { const now = nowInSeconds(); db.prepare("UPDATE streams SET archived_at = ? WHERE id = ?").run(now, id); + await invalidateCache("streams:list:"); + await invalidateCache("streams:export:"); + return true; } \ No newline at end of file diff --git a/frontend/src/pages/DashboardPage.tsx b/frontend/src/pages/DashboardPage.tsx index 7404ae3d..75acd598 100644 --- a/frontend/src/pages/DashboardPage.tsx +++ b/frontend/src/pages/DashboardPage.tsx @@ -16,6 +16,7 @@ import { ApiError, cancelStream, createStream, + fetchStats, getWebSocketUrl, listOpenIssues, listStreams, @@ -26,6 +27,19 @@ import { import { ListStreamsFilters } from "../services/api"; import { OpenIssue, Stream } from "../types/stream"; +interface PlatformStats { + total_streams: number; + active_streams: number; + completed_streams: number; + canceled_streams: number; + total_vested_by_asset: { + USDC: number; + XLM: number; + }; + unique_senders: number; + unique_recipients: number; +} + export interface DashboardPageProps { wallet?: FreighterState; } @@ -38,6 +52,7 @@ export function DashboardPage({ wallet: propWallet }: DashboardPageProps) { const [detailStreamId, setDetailStreamId] = useState(null); const [streams, setStreams] = useState([]); const [issues, setIssues] = useState([]); + const [platformStats, setPlatformStats] = useState(null); const [formError, setFormError] = useState(null); const [editingStream, setEditingStream] = useState<{ stream: Stream; @@ -49,14 +64,15 @@ export function DashboardPage({ wallet: propWallet }: DashboardPageProps) { const CREATE_STREAM_SECTION_ID = "create-stream-section"; const scrollToCreateStream = useCallback(() => { - document.getElementById(CREATE_STREAM_SECTION_ID)?.scrollIntoView({ behavior: "smooth" }); + document + .getElementById(CREATE_STREAM_SECTION_ID) + ?.scrollIntoView({ behavior: "smooth" }); }, []); const { filters, filteredStreams, setFilter } = useStreamFilter(streams); const [hasMore, setHasMore] = useState(true); const [loadingMore, setLoadingMore] = useState(false); const [currentPage, setCurrentPage] = useState(1); - const wsUrl = getWebSocketUrl(); const { lastMessage } = useWebSocket<{ eventType?: string; @@ -68,21 +84,15 @@ export function DashboardPage({ wallet: propWallet }: DashboardPageProps) { }>(wsUrl); const metricsHistory = useMetricsHistory("7d"); - const metrics = useMemo( - () => { - const active = streams.filter((stream) => stream.progress.status === "active").length; - const completed = streams.filter((stream) => stream.progress.status === "completed").length; - const vested = streams.reduce((sum, stream) => sum + stream.progress.vestedAmount, 0); - - return { - total: totalUnfilteredCount, - active, - completed, - vested, - }; - }, - [streams, totalUnfilteredCount], + () => ({ + total: platformStats?.total_streams ?? totalUnfilteredCount, + active: platformStats?.active_streams ?? 0, + completed: platformStats?.completed_streams ?? 0, + vestedUSDC: platformStats?.total_vested_by_asset.USDC ?? 0, + vestedXLM: platformStats?.total_vested_by_asset.XLM ?? 0, + }), + [platformStats, totalUnfilteredCount], ); const apiFilters: ListStreamsFilters = useMemo( @@ -108,7 +118,9 @@ export function DashboardPage({ wallet: propWallet }: DashboardPageProps) { [filters], ); - async function refreshStreams(currentFilters: ListStreamsFilters): Promise { + async function refreshStreams( + currentFilters: ListStreamsFilters, + ): Promise { const result = await listStreams({ ...currentFilters, limit: 20 }); setStreams(result.data); setHasMore(result.page * result.limit < result.total); @@ -120,7 +132,11 @@ export function DashboardPage({ wallet: propWallet }: DashboardPageProps) { setLoadingMore(true); try { const nextPage = currentPage + 1; - const result = await listStreams({ ...apiFilters, page: nextPage, limit: 20 }); + const result = await listStreams({ + ...apiFilters, + page: nextPage, + limit: 20, + }); setStreams((prev) => [...prev, ...result.data]); setHasMore(result.page * result.limit < result.total); setCurrentPage(result.page); @@ -138,8 +154,18 @@ export function DashboardPage({ wallet: propWallet }: DashboardPageProps) { } } + async function refreshPlatformStats(): Promise { + try { + const stats = (await fetchStats()) as unknown as PlatformStats; + setPlatformStats(stats); + } catch { + // Keep the dashboard usable if the public stats endpoint is unavailable. + } + } + useEffect(() => { void refreshUnfilteredCount(); + void refreshPlatformStats(); }, []); useEffect(() => { @@ -177,8 +203,11 @@ export function DashboardPage({ wallet: propWallet }: DashboardPageProps) { ...stream, progress: { ...stream.progress, - vestedAmount: lastMessage.vestedAmount ?? stream.progress.vestedAmount, - percentComplete: lastMessage.percentComplete ?? stream.progress.percentComplete, + vestedAmount: + lastMessage.vestedAmount ?? stream.progress.vestedAmount, + percentComplete: + lastMessage.percentComplete ?? + stream.progress.percentComplete, }, } : stream, @@ -195,6 +224,7 @@ export function DashboardPage({ wallet: propWallet }: DashboardPageProps) { void refreshStreams(apiFilters); void refreshUnfilteredCount(); + void refreshPlatformStats(); } }, [apiFilters, lastMessage, showToast]); @@ -206,6 +236,7 @@ export function DashboardPage({ wallet: propWallet }: DashboardPageProps) { await createStream(payload); await refreshStreams(apiFilters); void refreshUnfilteredCount(); + void refreshPlatformStats(); showToast("Stream created successfully", "success"); } catch (err) { if (err instanceof ApiError) { @@ -213,7 +244,9 @@ export function DashboardPage({ wallet: propWallet }: DashboardPageProps) { showToast(`Create failed (${err.statusCode}): ${err.message}`, "error"); return; } - const fallback = err instanceof Error ? err.message : "Failed to create stream."; + + const fallback = + err instanceof Error ? err.message : "Failed to create stream."; setFormError(fallback); showToast(fallback, "error"); } @@ -224,6 +257,7 @@ export function DashboardPage({ wallet: propWallet }: DashboardPageProps) { await cancelStream(streamId); await refreshStreams(apiFilters); void refreshUnfilteredCount(); + void refreshPlatformStats(); showToast("Stream canceled", "info"); } catch (err) { if (err instanceof ApiError) { @@ -243,13 +277,17 @@ export function DashboardPage({ wallet: propWallet }: DashboardPageProps) { await pauseStream(streamId); await refreshStreams(apiFilters); void refreshUnfilteredCount(); + void refreshPlatformStats(); showToast("Stream paused", "info"); } catch (err) { if (err instanceof ApiError) { showToast(`Pause failed (${err.statusCode}): ${err.message}`, "error"); return; } - showToast(err instanceof Error ? err.message : "Failed to pause the stream.", "error"); + showToast( + err instanceof Error ? err.message : "Failed to pause the stream.", + "error", + ); } } @@ -258,20 +296,28 @@ export function DashboardPage({ wallet: propWallet }: DashboardPageProps) { await resumeStream(streamId); await refreshStreams(apiFilters); void refreshUnfilteredCount(); + void refreshPlatformStats(); showToast("Stream resumed", "success"); } catch (err) { if (err instanceof ApiError) { showToast(`Resume failed (${err.statusCode}): ${err.message}`, "error"); return; } - showToast(err instanceof Error ? err.message : "Failed to resume the stream.", "error"); + showToast( + err instanceof Error ? err.message : "Failed to resume the stream.", + "error", + ); } } - async function handleUpdateStartTime(streamId: string, nextStartAt: number) { + async function handleUpdateStartTime( + streamId: string, + nextStartAt: number, + ) { try { await updateStreamStartAt(streamId, nextStartAt); await refreshStreams(apiFilters); + void refreshPlatformStats(); showToast("Start time updated", "success"); } catch (err) { if (err instanceof ApiError) { @@ -304,7 +350,9 @@ export function DashboardPage({ wallet: propWallet }: DashboardPageProps) {
Total Vested - {metrics.vested} + + {metrics.vestedUSDC} USDC / {metrics.vestedXLM} XLM +
@@ -325,6 +373,7 @@ export function DashboardPage({ wallet: propWallet }: DashboardPageProps) { walletAddress={wallet.address} /> + { setFilter("status", next.status ?? defaultStreamFilters.status); setFilter("sender", next.sender ?? defaultStreamFilters.sender); - setFilter("recipient", next.recipient ?? defaultStreamFilters.recipient); - setFilter("assetCode", next.asset ?? defaultStreamFilters.assetCode); + setFilter( + "recipient", + next.recipient ?? defaultStreamFilters.recipient, + ); + setFilter( + "assetCode", + next.asset ?? defaultStreamFilters.assetCode, + ); }} setUrlFilters={setUrlFilters} onCancel={handleCancel}