feat(ProofOfHeart-frontend): performance-uselivecampaignfunding-polls-every-n

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,20 @@ 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 +79,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);
+  });
+});

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<Set<string>>(new Set());
-  const cursorRef = useRef<string | undefined>(undefined);
   const onContributionsRef = useRef(onContributions);
   const queryClient = useQueryClient();
   const { publicKey: currentWalletAddress } = useWallet();
@@ -39,26 +36,16 @@ export function useCampaignContributionEvents({
 
   useEffect(() => {
     seenEventIdsRef.current = new Set();
-    cursorRef.current = undefined;
   }, [campaignId]);
 
   useEffect(() => {
     if (!enabled || !campaignId || USE_MOCKS || !isVisible) {
       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]);
 }

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);

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<typeof setTimeout> | 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.
  */