fix(stellar): make service-worker event pagination and cursor advancement lossless

e2e/scanner-retention.spec.ts

@@ -32,9 +32,16 @@ test('recovers an expired Stellar scanner cursor without a silent jump', async (
     const { default: ScannerWorker } =
       await import('/src/workers/stellar-scanner.worker.ts?worker');
     const worker = new ScannerWorker();
+    // The worker reports per-page progress before its terminal message, so wait
+    // for the terminal one (SUCCESS / ERROR / RETENTION_GAP).
     const scan = (startLedger: number) =>
       new Promise<Record<string, unknown>>((resolve, reject) => {
-        worker.onmessage = (event) => resolve(event.data);
+        worker.onmessage = (event) => {
+          const type = (event.data as { type?: string })?.type;
+          if (type === 'SUCCESS' || type === 'ERROR' || type === 'RETENTION_GAP') {
+            resolve(event.data);
+          }
+        };
         worker.onerror = (event) => reject(new Error(event.message));
         worker.postMessage({
           rpcUrl: 'https://soroban-testnet.stellar.org',
@@ -64,6 +71,7 @@ test('recovers an expired Stellar scanner cursor without a silent jump', async (
     oldestAvailableLedger: 250,
   });
   expect(result.cursorAfterError).toBe('100');
-  expect(result.recovered).toEqual({ type: 'SUCCESS', results: [], nextLedger: 501 });
+  // `recovered` also carries the committed boundary and page/truncation stats.
+  expect(result.recovered).toMatchObject({ type: 'SUCCESS', results: [], nextLedger: 501 });
   expect(result.finalCursor).toBe('501');
 });

src/components/StellarReceive.tsx

@@ -17,7 +17,6 @@ import {
   signStellarTransaction,
   STEALTH_SIGNING_MESSAGE,
   SCHEME_ID,
-  bytesToHex,
 } from '@wraith-protocol/sdk/chains/stellar';
 import type { Announcement, MatchedAnnouncement } from '@wraith-protocol/sdk/chains/stellar';
 import { useTranslation } from 'react-i18next';
@@ -44,122 +43,27 @@ import { useStealthLabels } from '@/hooks/useStealthLabels';
 import { StellarBatchWithdrawModal } from '@/components/StellarBatchWithdrawModal';
 import { createStellarQrUri } from '@/utils/qr';
 import { readScanCursor, writeScanCursor, type RetentionGap } from '@/lib/stellar/scannerCursor';
+import { fetchAnnouncementEvents } from '@/lib/stellar/announcementRpc';
 
 const ANNOUNCER_CONTRACT = 'CCJLJ2QRBJAAKIG6ELNQVXLLWMKKWVN5O2FKWUETHZGMPAD4MHK7WVWL';
 const REGISTRY_CONTRACT = 'CC2LAUCXYOPJ4DV4CYXNXYAXRDVOTMAWFF76W4WFD5OVQBD6TN4PYYJ5';
 
 // Fetches announcements on the main thread. Only used to support the
 // `window.scanAnnouncementsMock` test hook (see e2e/fixtures.ts) which needs
-// synchronous, main-thread access to the parsed announcements. The real,
-// production scan path fetches and scans off the main thread inside
-// stellar-scanner.worker.ts instead.
-async function fetchAnnouncementEvents(
+// main-thread access to the parsed announcements. The real, production scan
+// path fetches and scans off the main thread inside stellar-scanner.worker.ts
+// and shares this module's pagination and retention handling.
+async function fetchAnnouncementsBestEffort(
   rpcUrl: string,
   contractId: string,
 ): Promise<Announcement[]> {
-  const all: Announcement[] = [];
-
   try {
-    let startLedger = 1;
-    const probeRes = await fetch(rpcUrl, {
-      method: 'POST',
-      headers: { 'Content-Type': 'application/json' },
-      body: JSON.stringify({
-        jsonrpc: '2.0',
-        id: 0,
-        method: 'getEvents',
-        params: {
-          startLedger: 1,
-          filters: [{ type: 'contract', contractIds: [contractId] }],
-          pagination: { limit: 1 },
-        },
-      }),
-    });
-    const probeData = await probeRes.json();
-
-    if (probeData.error?.message) {
-      const match = probeData.error.message.match(/range:\s*(\d+)\s*-\s*(\d+)/);
-      if (match) {
-        const oldest = parseInt(match[1], 10);
-        const latest = parseInt(match[2], 10);
-        startLedger = Math.max(oldest, latest - 5000);
-      } else {
-        return all;
-      }
-    }
-
-    let cursor: string | undefined;
-    let hasMore = true;
-
-    while (hasMore) {
-      const params: Record<string, unknown> = {
-        filters: [{ type: 'contract', contractIds: [contractId] }],
-        pagination: { limit: 1000 },
-      };
-
-      if (cursor) {
-        (params.pagination as Record<string, unknown>).cursor = cursor;
-      } else {
-        params.startLedger = startLedger;
-      }
-
-      const res = await fetch(rpcUrl, {
-        method: 'POST',
-        headers: { 'Content-Type': 'application/json' },
-        body: JSON.stringify({ jsonrpc: '2.0', id: 2, method: 'getEvents', params }),
-      });
-
-      const data = await res.json();
-      const events = data.result?.events ?? [];
-
-      for (const event of events) {
-        try {
-          const ann = parseAnnouncementEvent(event);
-          if (ann) all.push(ann);
-        } catch (err) {
-          console.error('Failed to parse announcement event:', err);
-        }
-      }
-
-      if (events.length < 1000) {
-        hasMore = false;
-      } else {
-        cursor = data.result?.cursor;
-        if (!cursor) hasMore = false;
-      }
-    }
+    const { announcements } = await fetchAnnouncementEvents({ rpcUrl, contractId });
+    return announcements;
   } catch {
-    // Events API may not be available
+    // The events API may be unavailable; the mock hook tolerates an empty list.
+    return [];
   }
-
-  return all;
-}
-
-function parseAnnouncementEvent(event: Record<string, unknown>): Announcement | null {
-  const topics = event.topic as string[];
-  if (!topics || topics.length < 3) return null;
-
-  const schemeIdScVal = xdr.ScVal.fromXDR(topics[1], 'base64');
-  const schemeId = schemeIdScVal.u32();
-
-  const stealthScVal = xdr.ScVal.fromXDR(topics[2], 'base64');
-  const stealthScAddress = stealthScVal.address();
-  const stealthAddress = Address.fromScAddress(stealthScAddress).toString();
-
-  const valueScVal = xdr.ScVal.fromXDR(event.value as string, 'base64');
-  const valueVec = valueScVal.vec();
-  if (!valueVec || valueVec.length < 3) return null;
-
-  const callerScAddress = valueVec[0].address();
-  const caller = Address.fromScAddress(callerScAddress).toString();
-
-  const ephBytes = valueVec[1].bytes();
-  const ephemeralPubKey = bytesToHex(new Uint8Array(ephBytes));
-
-  const metaBytes = valueVec[2].bytes();
-  const metadata = bytesToHex(new Uint8Array(metaBytes));
-
-  return { schemeId, stealthAddress, caller, ephemeralPubKey, metadata };
 }
 
 function StellarMatchCardContainer({
@@ -1200,7 +1104,7 @@ export function StellarReceive() {
       const mockScan = (window as any).scanAnnouncementsMock;
       if (mockScan) {
         try {
-          const announcements = await fetchAnnouncementEvents(
+          const announcements = await fetchAnnouncementsBestEffort(
             STELLAR_NETWORK.rpcUrl,
             ANNOUNCER_CONTRACT,
           );
@@ -1239,10 +1143,29 @@ export function StellarReceive() {
           type,
           results,
           nextLedger,
+          committedLedger,
+          matched: pageMatches,
           requestedLedger,
           oldestAvailableLedger,
           error: workerError,
         } = event.data ?? {};
+        if (type === 'PAGE') {
+          // Each page advances the stored cursor only once its matches have been
+          // handed to the page, so an interrupted scan resumes at the first
+          // ledger proven complete instead of at the tip. A ledger that the RPC
+          // page boundary split is re-read, and matches are deduped below.
+          if (address && Number.isSafeInteger(committedLedger)) {
+            writeScanCursor(localStorage, address, committedLedger + 1);
+          }
+          if (Array.isArray(pageMatches) && pageMatches.length > 0) {
+            setMatched((previous) => {
+              const seen = new Set(previous.map((match) => match.stealthAddress));
+              const fresh = pageMatches.filter((match) => !seen.has(match.stealthAddress));
+              return fresh.length > 0 ? [...previous, ...fresh] : previous;
+            });
+          }
+          return;
+        }
         if (type === 'SUCCESS') {
           setMatched(results ?? []);
           setHasScanned(true);

src/hooks/useNotificationSW.ts

@@ -6,6 +6,7 @@ import {
   testRelayConnectivity,
   DEFAULT_RELAY_URL,
 } from '@/lib/pushRelay';
+import { getAppServiceWorkerRegistration } from '@/lib/swRegistration';
 
 interface SWMessage {
   type: string;
@@ -99,10 +100,10 @@ export function useNotificationSW(): UseNotificationSWReturn {
 
     async function registerSW() {
       try {
-        const registration = await navigator.serviceWorker.register(
-          new URL('../sw/stellar-notification-sw.ts', import.meta.url),
-          { type: 'module' },
-        );
+        // Reuse the unified app worker instead of registering a second script at
+        // the same scope, which would replace the PWA registration.
+        const registration = await getAppServiceWorkerRegistration();
+        if (!registration) throw new Error('App service worker is not registered');
 
         if (cancelled) return;
 

src/hooks/useStellarNotifications.ts

@@ -11,8 +11,8 @@
 
 import { useState, useEffect, useCallback, useRef } from 'react';
 import type { StealthKeys } from '@wraith-protocol/sdk/chains/stellar';
+import { getAppServiceWorkerRegistration } from '@/lib/swRegistration';
 
-const SW_PATH = '/sw/stellar-notification-sw.js';
 const STORAGE_KEY_OPT_IN = 'wraith:stellar:notifications-opt-in';
 const DB_NAME = 'wraith-stellar-notifications';
 const DB_VERSION = 1;
@@ -73,17 +73,16 @@ export function useStellarNotifications(): UseStellarNotificationsReturn {
     setState((prev: NotificationState) => ({ ...prev, enabled: optIn }));
   }, []);
 
-  // Register service worker
+  // Attach to the app's unified service worker (registered by useAppUpdate).
   useEffect(() => {
     if (!state.supported || !state.enabled) return;
 
     let cancelled = false;
 
     async function registerSW() {
       try {
-        const registration = await navigator.serviceWorker.register(SW_PATH, {
-          type: 'module',
-        });
+        const registration = await getAppServiceWorkerRegistration();
+        if (!registration) throw new Error('App service worker is not registered');
 
         if (cancelled) return;
 

src/lib/stellar/announcementRpc.ts

@@ -0,0 +1,126 @@
+/**
+ * Announcement-level view over the lossless `getEvents` pagination in
+ * `rpcEvents.ts`: it decodes raw `ScVal` topics/values into SDK
+ * `Announcement`s and reports progress per page so scanners can persist a
+ * cursor that only ever covers events they have already seen.
+ */
+import { Address, xdr } from '@stellar/stellar-sdk';
+import { bytesToHex } from '@wraith-protocol/sdk/chains/stellar';
+import type { Announcement } from '@wraith-protocol/sdk/chains/stellar';
+import { fetchEventPages, type StellarRpcEvent, type StellarRpcOptions } from './rpcEvents';
+
+export interface AnnouncementPage {
+  /** 1-based page index within this scan. */
+  page: number;
+  announcements: Announcement[];
+  /** Events whose announcement payload could not be decoded. */
+  undecodableEvents: number;
+  /**
+   * Highest ledger proven fully consumed by processed pages so far
+   * (`committedLedger + 1` resumes). A ledger the RPC cursor stopped inside is
+   * excluded, so its remaining announcements are re-read on resume.
+   */
+  committedLedger: number;
+}
+
+export interface FetchAnnouncementEventsOptions extends StellarRpcOptions {
+  rpcUrl: string;
+  contractId: string;
+  /** Ledger to resume from. Defaults to the edge of the retention window. */
+  startLedger?: number;
+  /** Ledger the scan must not read past. Defaults to the tip at scan start. */
+  throughLedger?: number;
+  pageLimit?: number;
+  lookbackLedgers?: number;
+  maxPages?: number;
+  /** Awaited after every page that has been decoded for the caller. */
+  onPage?: (page: AnnouncementPage) => void | Promise<void>;
+}
+
+export interface FetchAnnouncementEventsResult {
+  announcements: Announcement[];
+  /** Ledger the next scan resumes from: one past `committedLedger`. */
+  nextLedger: number;
+  committedLedger: number;
+  pages: number;
+  /** True when the page safety stop was hit before the boundary. */
+  truncated: boolean;
+}
+
+/**
+ * Decodes one `getEvents` entry into an `Announcement`. Returns `null` for
+ * entries that are not announcer announcements so a single malformed payload
+ * cannot abort a scan, and never mutates the cursor.
+ */
+export function parseAnnouncementEvent(
+  event: StellarRpcEvent | Record<string, unknown>,
+): Announcement | null {
+  const topics = event.topic as unknown;
+  if (!Array.isArray(topics) || topics.length < 3) return null;
+  if (typeof event.value !== 'string') return null;
+
+  const schemeIdScVal = xdr.ScVal.fromXDR(topics[1] as string, 'base64');
+  const schemeId = schemeIdScVal.u32();
+
+  const stealthScVal = xdr.ScVal.fromXDR(topics[2] as string, 'base64');
+  const stealthAddress = Address.fromScAddress(stealthScVal.address()).toString();
+
+  const valueVec = xdr.ScVal.fromXDR(event.value, 'base64').vec();
+  if (!valueVec || valueVec.length < 3) return null;
+
+  const caller = Address.fromScAddress(valueVec[0].address()).toString();
+  const ephemeralPubKey = bytesToHex(new Uint8Array(valueVec[1].bytes()));
+  const metadata = bytesToHex(new Uint8Array(valueVec[2].bytes()));
+
+  return { schemeId, stealthAddress, caller, ephemeralPubKey, metadata };
+}
+
+/**
+ * Reads every announcement the announcer contract emitted through the scan
+ * boundary, streaming decoded pages to `onPage`.
+ */
+export async function fetchAnnouncementEvents(
+  options: FetchAnnouncementEventsOptions,
+): Promise<FetchAnnouncementEventsResult> {
+  const announcements: Announcement[] = [];
+
+  const result = await fetchEventPages({
+    ...options,
+    onPage: async (page) => {
+      const pageAnnouncements: Announcement[] = [];
+      let undecodableEvents = 0;
+
+      for (const event of page.events) {
+        let announcement: Announcement | null = null;
+        try {
+          announcement = parseAnnouncementEvent(event);
+        } catch {
+          // Undecodable payloads are reported, not fatal: the cursor must stay
+          // lossless for the events that do decode.
+          announcement = null;
+        }
+        if (announcement) {
+          pageAnnouncements.push(announcement);
+        } else {
+          undecodableEvents += 1;
+        }
+      }
+
+      announcements.push(...pageAnnouncements);
+      await options.onPage?.({
+        page: page.page,
+        announcements: pageAnnouncements,
+        undecodableEvents,
+        committedLedger: page.committedLedger,
+      });
+    },
+  });
+
+  return {
+    announcements,
+    nextLedger: result.nextLedger,
+    committedLedger: result.committedLedger,
+    pages: result.pages,
+    truncated: result.truncated,
+  };
+}