From 4b954784e523197afb02852ada1137ea8ee6e954 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E2=80=9Ccaneryy=E2=80=9D?= Date: Tue, 30 Jun 2026 09:31:23 +0300 Subject: [PATCH 1/2] perf(funding): stream contribution_made events via Soroban cursor --- src/__tests__/lib/sorobanEvents.test.ts | 122 ++++++++++++++++++++- src/hooks/useCampaignContributionEvents.ts | 42 ++----- src/hooks/useLiveCampaignFunding.ts | 4 +- src/lib/sorobanEvents.ts | 90 +++++++++++++++ 4 files changed, 224 insertions(+), 34 deletions(-) diff --git a/src/__tests__/lib/sorobanEvents.test.ts b/src/__tests__/lib/sorobanEvents.test.ts index d7d17a96..78a4028d 100644 --- a/src/__tests__/lib/sorobanEvents.test.ts +++ b/src/__tests__/lib/sorobanEvents.test.ts @@ -4,10 +4,20 @@ jest.mock("@stellar/stellar-sdk", () => { toXDR: () => Buffer.from(`${opts.type}:${String(value)}`), }); + const mockGetLatestLedger = jest.fn().mockResolvedValue({ sequence: 1000 }); + const mockGetEvents = jest.fn(); + + class MockServer { + getLatestLedger = mockGetLatestLedger; + getEvents = mockGetEvents; + } + return { nativeToScVal, scValToNative: (value: { __native?: unknown }) => value.__native, - rpc: { Server: jest.fn() }, + rpc: { Server: MockServer }, + __mockGetEvents: mockGetEvents, + __mockGetLatestLedger: mockGetLatestLedger, }; }); @@ -20,6 +30,22 @@ import { sumContributionAmounts, } from "@/lib/sorobanEvents"; +const mockGetEvents = ( + StellarSdk as unknown as { __mockGetEvents: jest.Mock } +).__mockGetEvents; + +function makeContributionEvent(id: string, campaignId: number, amount: bigint) { + return { + id, + topic: [ + StellarSdk.nativeToScVal("contribution_made", { type: "symbol" }), + StellarSdk.nativeToScVal(campaignId, { type: "u32" }), + StellarSdk.nativeToScVal("GABC", { type: "address" }), + ], + value: { __bigint: amount }, + }; +} + describe("sorobanEvents vote cast", () => { it("builds campaign_vote_cast topic filter segments", () => { const topics = voteCastTopicFilter(7); @@ -55,3 +81,97 @@ describe("sorobanEvents vote cast", () => { expect(sumContributionAmounts([event, event])).toBe(BigInt(5_000_000)); }); }); + +describe("subscribeContributionMadeEvents", () => { + const originalEnv = process.env; + + beforeEach(() => { + jest.useFakeTimers(); + jest.clearAllMocks(); + jest.resetModules(); + process.env = { + ...originalEnv, + NEXT_PUBLIC_CONTRACT_ADDRESS: "CABC123", + NEXT_PUBLIC_USE_MOCKS: "false", + }; + }); + + afterEach(() => { + jest.useRealTimers(); + process.env = originalEnv; + }); + + async function loadSubscribe() { + const mod = await import("@/lib/sorobanEvents"); + return mod.subscribeContributionMadeEvents; + } + + it("returns null when event streaming is unavailable", async () => { + process.env = { + ...originalEnv, + NEXT_PUBLIC_USE_MOCKS: "true", + }; + jest.resetModules(); + + const subscribeContributionMadeEvents = await loadSubscribe(); + const subscription = subscribeContributionMadeEvents({ + campaignId: 1, + onEvents: jest.fn(), + }); + + expect(subscription).toBeNull(); + }); + + it("streams events via cursor and idles between empty polls", async () => { + const onEvents = jest.fn(); + mockGetEvents + .mockResolvedValueOnce({ + events: [makeContributionEvent("evt-1", 7, BigInt(1_000_000))], + cursor: "cursor-1", + latestLedger: 1000, + }) + .mockResolvedValueOnce({ + events: [], + cursor: "cursor-1", + latestLedger: 1000, + }); + + const subscribeContributionMadeEvents = await loadSubscribe(); + const subscription = subscribeContributionMadeEvents({ + campaignId: 7, + onEvents, + idleIntervalMs: 5000, + }); + + await Promise.resolve(); + expect(onEvents).toHaveBeenCalledTimes(1); + expect(onEvents.mock.calls[0][0].events).toHaveLength(1); + + await jest.advanceTimersByTimeAsync(5000); + expect(mockGetEvents).toHaveBeenCalledTimes(2); + + subscription?.unsubscribe(); + }); + + it("unsubscribes and stops further getEvents calls", async () => { + mockGetEvents.mockResolvedValue({ + events: [], + cursor: "cursor-1", + latestLedger: 1000, + }); + + const subscribeContributionMadeEvents = await loadSubscribe(); + const subscription = subscribeContributionMadeEvents({ + campaignId: 7, + onEvents: jest.fn(), + idleIntervalMs: 1000, + }); + + await Promise.resolve(); + subscription?.unsubscribe(); + + const callsBefore = mockGetEvents.mock.calls.length; + await jest.advanceTimersByTimeAsync(5000); + expect(mockGetEvents.mock.calls.length).toBe(callsBefore); + }); +}); diff --git a/src/hooks/useCampaignContributionEvents.ts b/src/hooks/useCampaignContributionEvents.ts index 2badd774..106421f6 100644 --- a/src/hooks/useCampaignContributionEvents.ts +++ b/src/hooks/useCampaignContributionEvents.ts @@ -1,14 +1,12 @@ "use client"; import { useEffect, useRef } from "react"; -import { fetchContributionMadeEvents, sumContributionAmounts } from "../lib/sorobanEvents"; +import { subscribeContributionMadeEvents, sumContributionAmounts } from "../lib/sorobanEvents"; import { useWindowVisibility } from "./useWindowVisibility"; import { useQueryClient } from "@tanstack/react-query"; import { useWallet } from "@/components/WalletContext"; import { invalidateQueriesForEvents } from "@/lib/cacheInvalidation"; -const EVENT_POLL_INTERVAL = Number(process.env.NEXT_PUBLIC_CONTRIBUTION_EVENTS_POLL_MS) || 5_000; - const USE_MOCKS = typeof process !== "undefined" && process.env.NEXT_PUBLIC_USE_MOCKS === "true"; export interface UseCampaignContributionEventsOptions { @@ -18,7 +16,7 @@ export interface UseCampaignContributionEventsOptions { } /** - * Polls Soroban `contribution_made` events for a campaign and reports new amounts. + * Streams Soroban `contribution_made` events for a campaign and reports new amounts. * Deduplicates by event id so reconnects do not double-count. */ export function useCampaignContributionEvents({ @@ -28,7 +26,6 @@ export function useCampaignContributionEvents({ }: UseCampaignContributionEventsOptions): void { const isVisible = useWindowVisibility(); const seenEventIdsRef = useRef>(new Set()); - const cursorRef = useRef(undefined); const onContributionsRef = useRef(onContributions); const queryClient = useQueryClient(); const { publicKey: currentWalletAddress } = useWallet(); @@ -39,7 +36,6 @@ export function useCampaignContributionEvents({ useEffect(() => { seenEventIdsRef.current = new Set(); - cursorRef.current = undefined; }, [campaignId]); useEffect(() => { @@ -47,18 +43,9 @@ export function useCampaignContributionEvents({ return; } - let cancelled = false; - - const poll = async () => { - try { - const result = await fetchContributionMadeEvents({ - campaignId, - cursor: cursorRef.current, - }); - if (!result || cancelled) return; - - cursorRef.current = result.cursor; - + const subscription = subscribeContributionMadeEvents({ + campaignId, + onEvents: (result) => { const unseen = result.events.filter((event) => !seenEventIdsRef.current.has(event.id)); for (const event of unseen) { seenEventIdsRef.current.add(event.id); @@ -67,23 +54,16 @@ export function useCampaignContributionEvents({ if (unseen.length > 0) { const delta = sumContributionAmounts(unseen); onContributionsRef.current?.(delta, unseen.length); - - // Invalidate relevant queries for the new events invalidateQueriesForEvents(queryClient, unseen, currentWalletAddress); } - } catch { + }, + onError: () => { // RPC errors are non-fatal; reconciliation via get_campaign covers drift. - } - }; - - void poll(); - const intervalId = window.setInterval(() => { - void poll(); - }, EVENT_POLL_INTERVAL); + }, + }); return () => { - cancelled = true; - window.clearInterval(intervalId); + subscription?.unsubscribe(); }; - }, [campaignId, enabled, isVisible]); + }, [campaignId, enabled, isVisible, queryClient, currentWalletAddress]); } diff --git a/src/hooks/useLiveCampaignFunding.ts b/src/hooks/useLiveCampaignFunding.ts index 664ffafb..9d53b40c 100644 --- a/src/hooks/useLiveCampaignFunding.ts +++ b/src/hooks/useLiveCampaignFunding.ts @@ -14,8 +14,8 @@ export interface UseLiveCampaignFundingResult { } /** - * Campaign detail funding with live increments from `contribution_made` events - * and periodic reconciliation via `get_campaign`. + * Campaign detail funding with live increments from streamed `contribution_made` events + * and reconciliation when `get_campaign` refetches. */ export function useLiveCampaignFunding(campaignId: number): UseLiveCampaignFundingResult { const { campaign, isLoading, error, refetch } = useCampaign(campaignId); diff --git a/src/lib/sorobanEvents.ts b/src/lib/sorobanEvents.ts index c768c8ae..46281804 100644 --- a/src/lib/sorobanEvents.ts +++ b/src/lib/sorobanEvents.ts @@ -142,6 +142,96 @@ export async function fetchContributionMadeEvents( }; } +const DEFAULT_CONTRIBUTION_STREAM_IDLE_MS = 2_000; +const DEFAULT_CONTRIBUTION_STREAM_MAX_BACKOFF_MS = 60_000; +const CONTRIBUTION_EVENT_BATCH_LIMIT = 100; + +export interface SubscribeContributionMadeEventsOptions { + campaignId: number; + onEvents: (result: FetchVoteCastEventsResult) => void; + onError?: (error: unknown) => void; + /** Delay between polls when no new events (default 2000ms). */ + idleIntervalMs?: number; +} + +export interface ContributionEventSubscription { + unsubscribe: () => void; +} + +/** + * Cursor-based Soroban event stream for `contribution_made` on a campaign. + * Uses chained getEvents calls instead of a fixed setInterval poll loop. + */ +export function subscribeContributionMadeEvents( + options: SubscribeContributionMadeEventsOptions, +): ContributionEventSubscription | null { + if (!isEventStreamingAvailable()) { + return null; + } + + const { + campaignId, + onEvents, + onError, + idleIntervalMs = Number(process.env.NEXT_PUBLIC_CONTRIBUTION_EVENTS_POLL_MS) || + DEFAULT_CONTRIBUTION_STREAM_IDLE_MS, + } = options; + + let cancelled = false; + let cursor: string | undefined; + let timeoutId: ReturnType | null = null; + let backoffMs = idleIntervalMs; + + const schedule = (delayMs: number) => { + if (cancelled) return; + timeoutId = setTimeout(() => { + timeoutId = null; + void poll(); + }, delayMs); + }; + + const poll = async () => { + if (cancelled) return; + + try { + const result = await fetchContributionMadeEvents({ + campaignId, + cursor, + limit: CONTRIBUTION_EVENT_BATCH_LIMIT, + }); + if (!result || cancelled) return; + + cursor = result.cursor; + + if (result.events.length > 0) { + onEvents(result); + backoffMs = idleIntervalMs; + schedule(result.events.length >= CONTRIBUTION_EVENT_BATCH_LIMIT ? 0 : idleIntervalMs); + return; + } + + backoffMs = idleIntervalMs; + schedule(idleIntervalMs); + } catch (error) { + onError?.(error); + backoffMs = Math.min(Math.round(backoffMs * 1.5), DEFAULT_CONTRIBUTION_STREAM_MAX_BACKOFF_MS); + schedule(backoffMs); + } + }; + + void poll(); + + return { + unsubscribe: () => { + cancelled = true; + if (timeoutId !== null) { + clearTimeout(timeoutId); + timeoutId = null; + } + }, + }; +} + /** * Poll Soroban RPC for `campaign_vote_cast` events on the configured contract. */ From 360e554dddc7755e7d0892dafb7844ff606dc3f9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E2=80=9Ccaneryy=E2=80=9D?= Date: Mon, 6 Jul 2026 00:32:37 +0300 Subject: [PATCH 2/2] style(tests): fix prettier formatting in sorobanEvents test Co-authored-by: Cursor --- src/__tests__/lib/sorobanEvents.test.ts | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/src/__tests__/lib/sorobanEvents.test.ts b/src/__tests__/lib/sorobanEvents.test.ts index 78a4028d..ae5ecba5 100644 --- a/src/__tests__/lib/sorobanEvents.test.ts +++ b/src/__tests__/lib/sorobanEvents.test.ts @@ -30,9 +30,7 @@ import { sumContributionAmounts, } from "@/lib/sorobanEvents"; -const mockGetEvents = ( - StellarSdk as unknown as { __mockGetEvents: jest.Mock } -).__mockGetEvents; +const mockGetEvents = (StellarSdk as unknown as { __mockGetEvents: jest.Mock }).__mockGetEvents; function makeContributionEvent(id: string, campaignId: number, amount: bigint) { return {