diff --git a/e2e/scanner-retention.spec.ts b/e2e/scanner-retention.spec.ts index 487e249..7f63f60 100644 --- a/e2e/scanner-retention.spec.ts +++ b/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>((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'); }); diff --git a/src/components/StellarReceive.tsx b/src/components/StellarReceive.tsx index 63b9d4e..c5b3ca5 100644 --- a/src/components/StellarReceive.tsx +++ b/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 { - 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 = { - filters: [{ type: 'contract', contractIds: [contractId] }], - pagination: { limit: 1000 }, - }; - - if (cursor) { - (params.pagination as Record).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): 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); diff --git a/src/hooks/useNotificationSW.ts b/src/hooks/useNotificationSW.ts index eaf0693..b1b2c9f 100644 --- a/src/hooks/useNotificationSW.ts +++ b/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; diff --git a/src/hooks/useStellarNotifications.ts b/src/hooks/useStellarNotifications.ts index c19936e..c0225c8 100644 --- a/src/hooks/useStellarNotifications.ts +++ b/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,7 +73,7 @@ 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; @@ -81,9 +81,8 @@ export function useStellarNotifications(): UseStellarNotificationsReturn { 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; diff --git a/src/lib/stellar/announcementRpc.ts b/src/lib/stellar/announcementRpc.ts new file mode 100644 index 0000000..113625f --- /dev/null +++ b/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; +} + +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, +): 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 { + 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, + }; +} diff --git a/src/lib/stellar/rpcEvents.test.ts b/src/lib/stellar/rpcEvents.test.ts new file mode 100644 index 0000000..6c033ba --- /dev/null +++ b/src/lib/stellar/rpcEvents.test.ts @@ -0,0 +1,440 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { + StellarRpcError, + StellarRpcHttpError, + StellarRpcResponseError, + StellarRpcTimeoutError, + fetchEventPages, + fetchLatestLedger, + type EventsPage, + type StellarRpcEvent, +} from './rpcEvents'; +import { StellarRetentionExceededError } from './scannerCursor'; + +const RPC_URL = 'https://soroban-testnet.stellar.org'; +const CONTRACT_ID = 'CCJLJ2QRBJAAKIG6ELNQVXLLWMKKWVN5O2FKWUETHZGMPAD4MHK7WVWL'; + +type RequestBody = Record; + +function ok(result: unknown): Response { + return new Response(JSON.stringify({ jsonrpc: '2.0', id: 1, result })); +} + +function rpcError(message: string, code = -32602): Response { + return new Response(JSON.stringify({ jsonrpc: '2.0', id: 1, error: { code, message } })); +} + +function mockRpc(responder: (body: RequestBody) => Response | Promise): RequestBody[] { + const requests: RequestBody[] = []; + vi.stubGlobal( + 'fetch', + vi.fn(async (_url: string, init?: RequestInit) => { + const body = JSON.parse((init?.body as string) ?? '{}') as RequestBody; + requests.push(body); + return responder(body); + }), + ); + return requests; +} + +function event(ledger: number): StellarRpcEvent { + return { topic: ['AAAA', 'AAAA', 'AAAA'], value: 'AAAA', ledger }; +} + +function eventRange(from: number, to: number): StellarRpcEvent[] { + const events: StellarRpcEvent[] = []; + for (let ledger = from; ledger <= to; ledger += 1) events.push(event(ledger)); + return events; +} + +/** Data requests are the ones that ask for a full page rather than probing. */ +function isDataRequest(body: RequestBody): boolean { + return body.method === 'getEvents' && body.params?.pagination?.limit !== 1; +} + +function isProbe(body: RequestBody): boolean { + return body.method === 'getEvents' && body.params?.pagination?.limit === 1; +} + +function pagedResponder(options: { + pages: StellarRpcEvent[][]; + latestLedger: number; + /** Soroban rejects requests older than this window. */ + retentionWindow?: { oldest: number; latest: number }; + /** 1-based data page that should fail. */ + failDataPage?: { page: number; status: number }; +}) { + let dataPage = 0; + return (body: RequestBody): Response => { + if (body.method === 'getLatestLedger') return ok({ sequence: options.latestLedger }); + if (body.method !== 'getEvents') return ok({}); + + const startLedger = body.params?.startLedger; + if ( + options.retentionWindow && + typeof startLedger === 'number' && + startLedger < options.retentionWindow.oldest + ) { + return rpcError( + `startLedger must be within the ledger range: ${options.retentionWindow.oldest} - ${options.retentionWindow.latest}`, + ); + } + + if (isProbe(body)) return ok({ events: [], latestLedger: options.latestLedger }); + + dataPage += 1; + if (options.failDataPage && options.failDataPage.page === dataPage) { + return new Response('upstream unavailable', { status: options.failDataPage.status }); + } + + const events = options.pages[dataPage - 1] ?? []; + const hasMore = dataPage < options.pages.length; + return ok({ + events, + latestLedger: options.latestLedger, + ...(hasMore ? { cursor: `c${dataPage}` } : {}), + }); + }; +} + +describe('Stellar RPC event pagination', () => { + afterEach(() => { + vi.unstubAllGlobals(); + }); + + it('follows pagination past the 1,000-event page cap and commits after every page', async () => { + const requests = mockRpc( + pagedResponder({ + latestLedger: 3_000, + pages: [eventRange(1, 1_000), eventRange(1_001, 2_000), eventRange(2_001, 2_500)], + }), + ); + const progress: EventsPage[] = []; + + const result = await fetchEventPages({ + rpcUrl: RPC_URL, + contractId: CONTRACT_ID, + onPage: (page) => { + progress.push(page); + }, + }); + + expect(result.events).toHaveLength(2_500); + expect(result.pages).toBe(3); + expect(result.truncated).toBe(false); + // A page boundary can split a ledger, so only ledgers below the boundary + // ledger are committed mid-scan; the final, exhausted page commits the tip. + expect(progress.map((page) => page.committedLedger)).toEqual([999, 1_999, 3_000]); + expect(progress.map((page) => page.events.length)).toEqual([1_000, 1_000, 500]); + + const dataRequests = requests.filter(isDataRequest); + expect(dataRequests[0].params.startLedger).toBe(1); + expect(dataRequests[1].params.pagination.cursor).toBe('c1'); + expect(dataRequests[1].params.startLedger).toBeUndefined(); + expect(dataRequests[2].params.pagination.cursor).toBe('c2'); + expect(result.nextLedger).toBe(3_001); + expect(result.committedLedger).toBe(3_000); + }); + + it('resumes from the last committed page after a mid-scan failure', async () => { + mockRpc( + pagedResponder({ + latestLedger: 3_000, + pages: [eventRange(1, 1_000), eventRange(1_001, 2_000), eventRange(2_001, 2_500)], + failDataPage: { page: 2, status: 503 }, + }), + ); + const progress: number[] = []; + + await expect( + fetchEventPages({ + rpcUrl: RPC_URL, + contractId: CONTRACT_ID, + startLedger: 1, + onPage: (page) => { + progress.push(page.committedLedger); + }, + }), + ).rejects.toBeInstanceOf(StellarRpcHttpError); + + // Only the first page was delivered, so only the first page may be committed. + expect(progress).toEqual([999]); + + // A restart from that committed cursor must read the failed page again + // instead of jumping to the tip. + const resumedRequests = mockRpc( + pagedResponder({ + latestLedger: 3_000, + pages: [eventRange(1_001, 2_000), eventRange(2_001, 2_500)], + }), + ); + const resumed = await fetchEventPages({ + rpcUrl: RPC_URL, + contractId: CONTRACT_ID, + startLedger: progress[progress.length - 1] + 1, + }); + + const dataRequests = resumedRequests.filter(isDataRequest); + expect(dataRequests[0].params.startLedger).toBe(1_000); + expect(dataRequests[1].params.pagination.cursor).toBe('c1'); + expect(resumed.events).toHaveLength(1_500); + expect(resumed.committedLedger).toBe(3_000); + expect(dataRequests).toHaveLength(2); + }); + + it('bounds every request with an abortable timeout', async () => { + vi.stubGlobal( + 'fetch', + vi.fn( + (_url: string, init?: RequestInit) => + new Promise((_resolve, reject) => { + init?.signal?.addEventListener('abort', () => + reject(new DOMException('The operation was aborted.', 'AbortError')), + ); + }), + ), + ); + const progress: EventsPage[] = []; + + await expect( + fetchEventPages({ + rpcUrl: RPC_URL, + contractId: CONTRACT_ID, + timeoutMs: 20, + onPage: (page) => { + progress.push(page); + }, + }), + ).rejects.toBeInstanceOf(StellarRpcTimeoutError); + + expect(progress).toEqual([]); + }); + + it('rejects non-2xx responses before trusting the body', async () => { + mockRpc(() => new Response('bad gateway', { status: 502 })); + + await expect( + fetchEventPages({ rpcUrl: RPC_URL, contractId: CONTRACT_ID }), + ).rejects.toMatchObject({ name: 'StellarRpcHttpError', status: 502 }); + }); + + it('validates the JSON-RPC envelope and result shape', async () => { + mockRpc(() => new Response('not json')); + await expect( + fetchEventPages({ rpcUrl: RPC_URL, contractId: CONTRACT_ID }), + ).rejects.toBeInstanceOf(StellarRpcResponseError); + + mockRpc(() => ok({ events: 'definitely-not-an-array' })); + await expect(fetchEventPages({ rpcUrl: RPC_URL, contractId: CONTRACT_ID })).rejects.toThrow( + /events was not an array/, + ); + + mockRpc(() => ok({})); + await expect(fetchLatestLedger(RPC_URL)).rejects.toThrow(/sequence/); + }); + + it('drops malformed event entries without losing the cursor', async () => { + mockRpc(() => + ok({ + events: [event(10), { value: 42 }, 'garbage', { topic: ['AAAA'] }], + latestLedger: 20, + }), + ); + + const result = await fetchEventPages({ + rpcUrl: RPC_URL, + contractId: CONTRACT_ID, + startLedger: 10, + throughLedger: 20, + }); + + expect(result.events).toHaveLength(1); + expect(result.malformedEvents).toBe(3); + expect(result.nextLedger).toBe(21); + }); + + it('maps an out-of-range cursor to a retention error carrying the window', async () => { + mockRpc( + pagedResponder({ + latestLedger: 500, + retentionWindow: { oldest: 250, latest: 500 }, + pages: [[]], + }), + ); + + await expect( + fetchEventPages({ rpcUrl: RPC_URL, contractId: CONTRACT_ID, startLedger: 100 }), + ).rejects.toBeInstanceOf(StellarRetentionExceededError); + + await expect( + fetchEventPages({ rpcUrl: RPC_URL, contractId: CONTRACT_ID, startLedger: 100 }), + ).rejects.toMatchObject({ requestedLedger: 100, oldestAvailableLedger: 250 }); + }); + + it('clamps a first scan to the retention window when no cursor is stored', async () => { + const requests = mockRpc( + pagedResponder({ + latestLedger: 500, + retentionWindow: { oldest: 250, latest: 500 }, + pages: [[]], + }), + ); + + const result = await fetchEventPages({ rpcUrl: RPC_URL, contractId: CONTRACT_ID }); + + const dataRequests = requests.filter(isDataRequest); + expect(dataRequests[0].params.startLedger).toBe(250); + expect(result.nextLedger).toBe(501); + expect(result.committedLedger).toBe(500); + }); + + it('stops at the committed boundary instead of chasing the tip', async () => { + const requests = mockRpc( + pagedResponder({ + latestLedger: 2_000, + pages: [eventRange(1, 10), [event(400)], [event(501)]], + }), + ); + + const result = await fetchEventPages({ + rpcUrl: RPC_URL, + contractId: CONTRACT_ID, + startLedger: 1, + throughLedger: 500, + }); + + expect(requests.filter(isDataRequest)).toHaveLength(3); + expect(result.committedLedger).toBe(500); + expect(result.nextLedger).toBe(501); + expect(result.truncated).toBe(false); + }); + + it('reports truncation instead of skipping pages when the page ceiling is hit', async () => { + mockRpc( + pagedResponder({ + latestLedger: 10_000, + pages: [eventRange(1, 100), eventRange(101, 200), eventRange(201, 300)], + }), + ); + const progress: number[] = []; + + const result = await fetchEventPages({ + rpcUrl: RPC_URL, + contractId: CONTRACT_ID, + startLedger: 1, + maxPages: 2, + onPage: (page) => { + progress.push(page.committedLedger); + }, + }); + + expect(result.truncated).toBe(true); + expect(result.pages).toBe(2); + expect(progress).toEqual([99, 199]); + // The next scan resumes at the last ledger actually proven complete. + expect(result.nextLedger).toBe(200); + }); + + it('does not commit a ledger whose events continue on a rejected page', async () => { + // Both pages carry events from ledger 500: page 1 stops inside it and page 2 + // was going to deliver the rest before failing. + mockRpc( + pagedResponder({ + latestLedger: 505, + pages: [ + [...eventRange(490, 499), event(500), event(500)], + [event(500), event(500)], + ], + failDataPage: { page: 2, status: 503 }, + }), + ); + const progress: EventsPage[] = []; + + await expect( + fetchEventPages({ + rpcUrl: RPC_URL, + contractId: CONTRACT_ID, + startLedger: 500, + onPage: (page) => { + progress.push(page); + }, + }), + ).rejects.toBeInstanceOf(StellarRpcHttpError); + + // Ledger 500 was only partly delivered before the page failed, so it must + // not be committed even though its events appeared on the first page. + expect(progress).toHaveLength(1); + expect(progress[0].events.filter((entry) => entry.ledger === 500)).toHaveLength(2); + expect(progress[0].committedLedger).toBe(499); + + const resumedRequests = mockRpc( + pagedResponder({ + latestLedger: 505, + pages: [ + [event(500), event(500)], + [event(500), event(500)], + ], + }), + ); + const resumedStartLedger = progress[progress.length - 1].committedLedger + 1; + const resumed = await fetchEventPages({ + rpcUrl: RPC_URL, + contractId: CONTRACT_ID, + startLedger: resumedStartLedger, + }); + + // The restart begins inside ledger 500 rather than at 501, so nothing is skipped. + expect(resumedStartLedger).toBe(500); + expect(resumedRequests.filter(isDataRequest)[0].params.startLedger).toBe(500); + expect(resumed.events.filter((entry) => entry.ledger === 500)).toHaveLength(4); + expect(resumed.committedLedger).toBe(505); + }); + + it('reads a boundary ledger to its end instead of stopping at the first page that touches it', async () => { + mockRpc( + pagedResponder({ + latestLedger: 500, + pages: [[event(500), event(500)], [event(500)]], + }), + ); + const progress: number[] = []; + + const result = await fetchEventPages({ + rpcUrl: RPC_URL, + contractId: CONTRACT_ID, + startLedger: 500, + throughLedger: 500, + onPage: (page) => { + progress.push(page.committedLedger); + }, + }); + + expect(result.events).toHaveLength(3); + expect(result.pages).toBe(2); + expect(progress).toEqual([499, 500]); + expect(result.committedLedger).toBe(500); + expect(result.nextLedger).toBe(501); + }); + + it('fails loudly when the RPC repeats a cursor', async () => { + let dataPage = 0; + mockRpc((body) => { + if (body.method === 'getLatestLedger') return ok({ sequence: 1_000 }); + if (isProbe(body)) return ok({ events: [], latestLedger: 1_000 }); + dataPage += 1; + return ok({ events: [event(dataPage * 10)], latestLedger: 1_000, cursor: 'stuck' }); + }); + + await expect( + fetchEventPages({ rpcUrl: RPC_URL, contractId: CONTRACT_ID, startLedger: 1 }), + ).rejects.toThrow(/cursor did not advance/); + }); + + it('surfaces JSON-RPC errors as typed errors', async () => { + mockRpc((body) => (isProbe(body) ? rpcError('boom', -32001) : ok({}))); + + await expect( + fetchEventPages({ rpcUrl: RPC_URL, contractId: CONTRACT_ID }), + ).rejects.toBeInstanceOf(StellarRpcError); + }); +}); diff --git a/src/lib/stellar/rpcEvents.ts b/src/lib/stellar/rpcEvents.ts new file mode 100644 index 0000000..2504372 --- /dev/null +++ b/src/lib/stellar/rpcEvents.ts @@ -0,0 +1,495 @@ +/** + * Lossless Soroban `getEvents` pagination over Stellar RPC. + * + * Every component that scans the announcer contract — the page-side scanner + * worker, the service worker's background sync and the SDK-backed scanner — + * reads events through this module so there is exactly one implementation of + * the cursor rules: + * + * 1. Pages are followed with the RPC's `pagination.cursor` until either the + * committed boundary or the end of the window is reached. The 1,000-event + * per-call cap is a page size, never a scan size. + * 2. Progress is persisted through `onPage`, which is awaited after every page + * that was successfully fetched *and* handed to the caller. A scan that + * dies mid-way therefore resumes from the first unread ledger instead of + * jumping to the tip. Because a page boundary can split a single ledger + * across two pages, the last ledger a page delivered is only committed once + * a later page proves it was read to the end. + * 3. Requests are abortable and time-bounded, HTTP status is checked before the + * body is read, and every JSON-RPC envelope/result is shape-validated before + * it is trusted. + * + * This module deliberately has no Stellar SDK imports so it can be bundled into + * the service worker without pulling in XDR parsing; decoding event payloads + * into announcements lives in `announcementRpc.ts`. + */ +import { + retentionErrorFromRpcMessage, + parseLedgerWindow, + StellarRetentionExceededError, +} from './scannerCursor'; + +/** Ceiling for a single RPC round-trip. */ +export const DEFAULT_RPC_TIMEOUT_MS = 15_000; +/** Page size requested from `getEvents`. The RPC caps responses at this value. */ +export const DEFAULT_EVENTS_PAGE_LIMIT = 1_000; +/** How far back a first scan reaches when the RPC serves the full history. */ +export const DEFAULT_EVENT_LOOKBACK_LEDGERS = 5_000; +/** Safety stop for pagination loops; a healthy scan needs far fewer pages. */ +export const MAX_EVENT_PAGES = 1_000; + +export class StellarRpcHttpError extends Error { + readonly status: number; + readonly method: string; + + constructor(rpcUrl: string, method: string, status: number, statusText: string) { + super( + `Stellar RPC ${method} at ${rpcUrl} failed with HTTP ${status}${statusText ? ` ${statusText}` : ''}`, + ); + this.name = 'StellarRpcHttpError'; + this.status = status; + this.method = method; + } +} + +export class StellarRpcTimeoutError extends Error { + readonly timeoutMs: number; + readonly method: string; + + constructor(rpcUrl: string, method: string, timeoutMs: number) { + super(`Stellar RPC ${method} at ${rpcUrl} timed out after ${timeoutMs}ms`); + this.name = 'StellarRpcTimeoutError'; + this.timeoutMs = timeoutMs; + this.method = method; + } +} + +export class StellarRpcResponseError extends Error { + readonly method: string; + + constructor(method: string, message: string) { + super(`Stellar RPC ${method} returned an invalid response: ${message}`); + this.name = 'StellarRpcResponseError'; + this.method = method; + } +} + +/** A JSON-RPC `error` object returned by the RPC with a 2xx status. */ +export class StellarRpcError extends Error { + readonly code?: number; + + constructor(message: string, code?: number) { + super(message); + this.name = 'StellarRpcError'; + this.code = code; + } +} + +/** A shape-validated `getEvents` entry. */ +export interface StellarRpcEvent { + /** Base64-encoded `ScVal` topics. */ + topic: string[]; + /** Base64-encoded `ScVal` value. */ + value: string; + /** Ledger the event was emitted in — the unit the scan cursor is kept in. */ + ledger?: number; + contractId?: string; +} + +export interface StellarRpcOptions { + /** Per-request timeout in milliseconds. Defaults to `DEFAULT_RPC_TIMEOUT_MS`. */ + timeoutMs?: number; + /** Caller-owned abort signal, combined with the per-request timeout. */ + signal?: AbortSignal; +} + +/** A fully processed page, handed to `onPage` before the next request is made. */ +export interface EventsPage { + /** 1-based page index within this scan. */ + page: number; + events: StellarRpcEvent[]; + /** Entries dropped because they were not a valid `getEvents` event. */ + malformedEvents: number; + /** + * Highest ledger proven fully consumed by this scan, including earlier pages. + * Safe to persist as `committedLedger + 1`: the ledger the RPC cursor stopped + * in is deliberately excluded while more events may still follow in it. + */ + committedLedger: number; +} + +export interface FetchEventPagesOptions 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; + /** + * RPC event filters. Defaults to every event from `contractId`; contracts that + * only care about specific topics pass their own filter list. + */ + filters?: unknown[]; + /** Page size. Defaults to `DEFAULT_EVENTS_PAGE_LIMIT`. */ + pageLimit?: number; + /** Look-back window for a first scan. Defaults to `DEFAULT_EVENT_LOOKBACK_LEDGERS`. */ + lookbackLedgers?: number; + /** Safety stop for the pagination loop. Defaults to `MAX_EVENT_PAGES`. */ + maxPages?: number; + /** + * Awaited after each successfully processed page. Persisting + * `committedLedger + 1` here is what makes an interrupted scan lossless; a + * resume may re-read the last ledger the page touched, but never skips past + * events that page had not delivered yet. + */ + onPage?: (page: EventsPage) => void | Promise; +} + +export interface FetchEventPagesResult { + events: StellarRpcEvent[]; + /** + * Ledger the next scan resumes from: one past `committedLedger`. It can point + * back into a ledger that was only partly delivered if the scan stopped there. + */ + nextLedger: number; + /** Highest ledger fully processed by this scan. */ + committedLedger: number; + /** Boundary this scan was clamped to. */ + throughLedger: number; + pages: number; + /** True when `maxPages` stopped the scan before the boundary. */ + truncated: boolean; + malformedEvents: number; +} + +interface GetEventsPage { + events: StellarRpcEvent[]; + cursor?: string; + latestLedger?: number; + malformedEvents: number; +} + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null; +} + +function validateEvent(raw: unknown): StellarRpcEvent | undefined { + if (!isRecord(raw)) return undefined; + if (typeof raw.value !== 'string') return undefined; + if (!Array.isArray(raw.topic) || raw.topic.some((topic) => typeof topic !== 'string')) { + return undefined; + } + + const event: StellarRpcEvent = { topic: raw.topic as string[], value: raw.value }; + if (typeof raw.ledger === 'number' && Number.isSafeInteger(raw.ledger) && raw.ledger > 0) { + event.ledger = raw.ledger; + } + if (typeof raw.contractId === 'string') { + event.contractId = raw.contractId; + } + return event; +} + +function validateEventsPage(result: unknown): GetEventsPage { + if (!isRecord(result)) { + throw new StellarRpcResponseError('getEvents', 'result was not an object'); + } + if (!Array.isArray(result.events)) { + throw new StellarRpcResponseError('getEvents', 'result.events was not an array'); + } + + const events: StellarRpcEvent[] = []; + let malformedEvents = 0; + for (const raw of result.events) { + const event = validateEvent(raw); + if (event) { + events.push(event); + } else { + malformedEvents += 1; + } + } + + const page: GetEventsPage = { events, malformedEvents }; + if (typeof result.cursor === 'string' && result.cursor.length > 0) { + page.cursor = result.cursor; + } + if (typeof result.latestLedger === 'number' && Number.isSafeInteger(result.latestLedger)) { + page.latestLedger = result.latestLedger; + } + return page; +} + +function unwrapJsonRpcResult(payload: unknown, method: string): unknown { + if (!isRecord(payload)) { + throw new StellarRpcResponseError(method, 'body was not a JSON object'); + } + if (payload.jsonrpc !== undefined && payload.jsonrpc !== '2.0') { + throw new StellarRpcResponseError(method, 'jsonrpc version was not "2.0"'); + } + if (payload.error !== undefined && payload.error !== null) { + const error = payload.error; + const message = + isRecord(error) && typeof error.message === 'string' + ? error.message + : `${method} rejected the request without a message`; + const code = + isRecord(error) && typeof error.code === 'number' ? (error.code as number) : undefined; + throw new StellarRpcError(message, code); + } + if (payload.result === undefined || payload.result === null) { + throw new StellarRpcResponseError(method, 'body did not contain a result'); + } + return payload.result; +} + +/** + * One validated, time-bounded JSON-RPC call. Network, HTTP, envelope and shape + * failures all surface as typed errors so callers can tell an outage (retry + * later, keep the cursor) from a retention gap (recover the cursor). + */ +async function postJsonRpc( + rpcUrl: string, + method: string, + params: Record | undefined, + options: StellarRpcOptions, +): Promise { + const timeoutMs = options.timeoutMs ?? DEFAULT_RPC_TIMEOUT_MS; + const controller = new AbortController(); + let timedOut = false; + const timer = setTimeout(() => { + timedOut = true; + controller.abort(); + }, timeoutMs); + const abortFromCaller = () => controller.abort(); + options.signal?.addEventListener('abort', abortFromCaller); + if (options.signal?.aborted) controller.abort(); + + try { + let response: Response; + try { + response = await fetch(rpcUrl, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ jsonrpc: '2.0', id: 1, method, ...(params ? { params } : {}) }), + signal: controller.signal, + }); + } catch (error) { + if (timedOut) throw new StellarRpcTimeoutError(rpcUrl, method, timeoutMs); + throw error; + } + + if (!response.ok) { + throw new StellarRpcHttpError(rpcUrl, method, response.status, response.statusText); + } + + let payload: unknown; + try { + payload = await response.json(); + } catch { + throw new StellarRpcResponseError(method, 'body was not valid JSON'); + } + + return unwrapJsonRpcResult(payload, method); + } finally { + clearTimeout(timer); + options.signal?.removeEventListener('abort', abortFromCaller); + } +} + +/** Reads the current tip. Throws when the RPC omits a usable sequence. */ +export async function fetchLatestLedger( + rpcUrl: string, + options: StellarRpcOptions = {}, +): Promise { + const result = await postJsonRpc(rpcUrl, 'getLatestLedger', undefined, options); + const sequence = isRecord(result) ? result.sequence : undefined; + if (typeof sequence !== 'number' || !Number.isSafeInteger(sequence) || sequence <= 0) { + throw new StellarRpcResponseError('getLatestLedger', 'result.sequence was not a ledger number'); + } + return sequence; +} + +function eventParams( + filters: unknown[], + pageLimit: number, + position: { startLedger: number } | { cursor: string }, +): Record { + const pagination: Record = { limit: pageLimit }; + if ('cursor' in position) pagination.cursor = position.cursor; + return { + ...('startLedger' in position ? { startLedger: position.startLedger } : {}), + filters, + pagination, + }; +} + +async function getEventsPage( + rpcUrl: string, + filters: unknown[], + pageLimit: number, + position: { startLedger: number } | { cursor: string }, + options: StellarRpcOptions, + /** When set, a range error is mapped to a retention error for this cursor. */ + retentionLedger?: number, +): Promise { + try { + const result = await postJsonRpc( + rpcUrl, + 'getEvents', + eventParams(filters, pageLimit, position), + options, + ); + return validateEventsPage(result); + } catch (error) { + if (retentionLedger !== undefined && error instanceof StellarRpcError) { + const retentionError = retentionErrorFromRpcMessage(retentionLedger, error.message); + if (retentionError) throw retentionError; + } + throw error; + } +} + +/** + * Probes the ledger window the RPC is willing to serve. Soroban rejects + * `startLedger: 1` once the window has moved past it and reports the available + * range in the error, which is the only way to learn the oldest readable + * ledger. + */ +async function probeLedgerWindow( + rpcUrl: string, + filters: unknown[], + options: StellarRpcOptions, +): Promise<{ oldest: number; latest?: number }> { + try { + const page = await getEventsPage( + rpcUrl, + filters, + 1, + { startLedger: 1 }, + options, + /* retentionLedger */ undefined, + ); + return { oldest: 1, latest: page.latestLedger }; + } catch (error) { + if (error instanceof StellarRpcError) { + const window = parseLedgerWindow(error.message); + if (window) return { oldest: window.oldest, latest: window.latest }; + } + throw error; + } +} + +/** + * Follows `getEvents` pagination until every event through the committed + * boundary has been read, calling `onPage` after each page. + */ +export async function fetchEventPages( + options: FetchEventPagesOptions, +): Promise { + const { rpcUrl, contractId, onPage } = options; + const rpcOptions: StellarRpcOptions = { timeoutMs: options.timeoutMs, signal: options.signal }; + const filters = options.filters ?? [{ type: 'contract', contractIds: [contractId] }]; + const pageLimit = options.pageLimit ?? DEFAULT_EVENTS_PAGE_LIMIT; + const lookbackLedgers = options.lookbackLedgers ?? DEFAULT_EVENT_LOOKBACK_LEDGERS; + const maxPages = options.maxPages ?? MAX_EVENT_PAGES; + + if (!Number.isSafeInteger(pageLimit) || pageLimit <= 0) { + throw new Error('pageLimit must be a positive integer'); + } + + const window = await probeLedgerWindow(rpcUrl, filters, rpcOptions); + const latest = window.latest ?? (await fetchLatestLedger(rpcUrl, rpcOptions)); + const oldest = window.oldest; + const startLedger = options.startLedger ?? Math.max(oldest, latest - lookbackLedgers); + if (!Number.isSafeInteger(startLedger) || startLedger <= 0) { + throw new Error('startLedger must be a positive integer'); + } + if (startLedger < oldest) { + throw new StellarRetentionExceededError(startLedger, oldest); + } + + const throughLedger = options.throughLedger ?? latest; + if (startLedger > throughLedger) { + return { + events: [], + nextLedger: startLedger, + committedLedger: startLedger - 1, + throughLedger, + pages: 0, + truncated: false, + malformedEvents: 0, + }; + } + + const events: StellarRpcEvent[] = []; + let committedLedger = startLedger - 1; + let malformedEvents = 0; + let cursor: string | undefined; + let pages = 0; + let truncated = false; + + while (pages < maxPages) { + const requestedLedger = committedLedger + 1; + const page = await getEventsPage( + rpcUrl, + filters, + pageLimit, + cursor ? { cursor } : { startLedger: requestedLedger }, + rpcOptions, + requestedLedger, + ); + pages += 1; + events.push(...page.events); + malformedEvents += page.malformedEvents; + + const pageMaxLedger = page.events.reduce( + (max, event) => + event.ledger !== undefined && (max === undefined || event.ledger > max) + ? event.ledger + : max, + undefined, + ); + + if (page.cursor === undefined) { + // The window is exhausted: every ledger the RPC was able to serve has + // been read, so the scan may commit up to the smaller of the two tips. + committedLedger = Math.min(throughLedger, page.latestLedger ?? throughLedger); + } else if (pageMaxLedger !== undefined) { + // A cursor means more events follow this page, so the page may have + // stopped partway through its last ledger. Events arrive in ledger order, + // which makes every ledger below that one fully consumed, while the last + // one is not: committing it would let a restart skip its remaining + // events. Commit up to the last proven ledger and re-read the boundary + // ledger on resume instead. + committedLedger = Math.max(committedLedger, Math.min(throughLedger, pageMaxLedger - 1)); + } + + await onPage?.({ + page: pages, + events: page.events, + malformedEvents: page.malformedEvents, + committedLedger, + }); + + if (page.cursor === undefined) break; + if (committedLedger >= throughLedger) break; + if (page.cursor === cursor) { + throw new StellarRpcResponseError('getEvents', 'cursor did not advance between pages'); + } + cursor = page.cursor; + } + + if (pages >= maxPages && cursor !== undefined && committedLedger < throughLedger) { + truncated = true; + } + + return { + events, + nextLedger: committedLedger + 1, + committedLedger, + throughLedger, + pages, + truncated, + malformedEvents, + }; +} diff --git a/src/lib/stellar/scannerCursor.ts b/src/lib/stellar/scannerCursor.ts index a5b4288..0e8b18d 100644 --- a/src/lib/stellar/scannerCursor.ts +++ b/src/lib/stellar/scannerCursor.ts @@ -63,13 +63,25 @@ export function retentionGapFromError(error: unknown): RetentionGap | undefined return { requestedLedger, oldestAvailableLedger }; } +/** + * Reads the ledger range Soroban reports when a request falls outside its + * retention window, e.g. `startLedger must be within the ledger range: 250 - 500`. + */ +export function parseLedgerWindow(message: string): { oldest: number; latest: number } | undefined { + const match = message.match(/range:\s*(\d+)\s*-\s*(\d+)/i); + if (!match) return undefined; + const oldest = Number(match[1]); + const latest = Number(match[2]); + if (!Number.isSafeInteger(oldest) || !Number.isSafeInteger(latest)) return undefined; + return { oldest, latest }; +} + export function retentionErrorFromRpcMessage( requestedLedger: number, message: string, ): StellarRetentionExceededError | undefined { - const match = message.match(/range:\s*(\d+)\s*-\s*(\d+)/i); - if (!match) return undefined; - const oldestAvailableLedger = Number(match[1]); - if (requestedLedger >= oldestAvailableLedger) return undefined; - return new StellarRetentionExceededError(requestedLedger, oldestAvailableLedger); + const window = parseLedgerWindow(message); + if (!window) return undefined; + if (requestedLedger >= window.oldest) return undefined; + return new StellarRetentionExceededError(requestedLedger, window.oldest); } diff --git a/src/lib/stellar/vaultClaim.test.ts b/src/lib/stellar/vaultClaim.test.ts index ccc3292..bdc149f 100644 --- a/src/lib/stellar/vaultClaim.test.ts +++ b/src/lib/stellar/vaultClaim.test.ts @@ -180,9 +180,19 @@ describe('fetchVaultDepositEvents', () => { vi.stubGlobal( 'fetch', - vi.fn(async () => ({ - json: async () => ({ result: { events: [event] } }), - })), + vi.fn(async (_url: string, init?: RequestInit) => { + const body = JSON.parse((init?.body as string) ?? '{}'); + if (body.method === 'getLatestLedger') { + return new Response( + JSON.stringify({ jsonrpc: '2.0', id: body.id, result: { sequence: 700 } }), + ); + } + // The retention probe asks for a single event; the scan asks for a page. + const events = body.params?.pagination?.limit === 1 ? [] : [event]; + return new Response( + JSON.stringify({ jsonrpc: '2.0', id: body.id, result: { events, latestLedger: 700 } }), + ); + }), ); const events = await fetchVaultDepositEvents( diff --git a/src/lib/stellar/vaultClaim.ts b/src/lib/stellar/vaultClaim.ts index 83e98b4..1c9d8ba 100644 --- a/src/lib/stellar/vaultClaim.ts +++ b/src/lib/stellar/vaultClaim.ts @@ -21,6 +21,7 @@ import { import type { Announcement } from '@wraith-protocol/sdk/chains/stellar'; import { STELLAR_NETWORK } from '@/config'; import { STELLAR_USDC } from '@/lib/stellar/assets'; +import { fetchEventPages, type StellarRpcEvent } from '@/lib/stellar/rpcEvents'; import { isBoundedString, isFiniteTimestamp, @@ -190,7 +191,9 @@ export interface VaultDepositEvent { unlockLedger: number; } -function parseVaultDepositEvent(event: Record): VaultDepositEvent | null { +function parseVaultDepositEvent( + event: StellarRpcEvent | Record, +): VaultDepositEvent | null { const topics = event.topic as string[]; if (!topics || topics.length < 2) return null; @@ -212,9 +215,10 @@ function parseVaultDepositEvent(event: Record): VaultDepositEve } /** - * Fetches every `deposit` event emitted by the stealth-vault contract via the - * Soroban RPC `getEvents` method. Mirrors the probe/pagination pattern already - * used for announcer events elsewhere in this codebase. + * Fetches every `deposit` event emitted by the stealth-vault contract. Reads go + * through the shared paginated `getEvents` implementation so the vault flow + * gets the same cursor, timeout and response-validation rules as the announcer + * scans. Still best-effort: an unavailable events API yields an empty list. */ export async function fetchVaultDepositEvents( rpcUrl: string, @@ -222,77 +226,29 @@ export async function fetchVaultDepositEvents( ): Promise { const all: VaultDepositEvent[] = []; const depositTopic = xdr.ScVal.scvSymbol('deposit').toXDR('base64'); - const contractFilter = { - type: 'contract', - contractIds: [vaultContractId], - topics: [[depositTopic, '*']], - }; 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: [contractFilter], 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 = { - filters: [contractFilter], - pagination: { limit: 1000 }, - }; - - if (cursor) { - (params.pagination as Record).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 parsed = parseVaultDepositEvent(event); - if (parsed) all.push(parsed); - } catch { - // Skip malformed + await fetchEventPages({ + rpcUrl, + contractId: vaultContractId, + filters: [ + { + type: 'contract', + contractIds: [vaultContractId], + topics: [[depositTopic, '*']], + }, + ], + onPage: (page) => { + for (const event of page.events) { + try { + const parsed = parseVaultDepositEvent(event); + if (parsed) all.push(parsed); + } catch { + // Skip malformed + } } - } - - if (events.length < 1000) { - hasMore = false; - } else { - cursor = data.result?.cursor; - if (!cursor) hasMore = false; - } - } + }, + }); } catch { // Events API may not be available } diff --git a/src/lib/swRegistration.ts b/src/lib/swRegistration.ts new file mode 100644 index 0000000..db77b20 --- /dev/null +++ b/src/lib/swRegistration.ts @@ -0,0 +1,40 @@ +/** + * Access to the single service worker the app runs. + * + * The PWA worker in `src/sw/app-sw.ts` owns every background responsibility — + * precaching, Stellar notification sync, viewing-key storage and push + * subscriptions — and is registered once per page load by `useAppUpdate` + * (`virtual:pwa-register`). Features that need to talk to it must reuse that + * registration: registering a second script at the same scope replaces the + * running worker, which used to leave the app with two competing scan paths and + * two independent cursors. + */ +const REGISTRATION_TIMEOUT_MS = 10_000; + +/** + * Resolves the app's service worker registration, or `undefined` when the + * browser has none (unsupported, or the worker is not registered yet). Waiting + * is bounded so a disabled/updated-out worker can never hang the caller. + */ +export async function getAppServiceWorkerRegistration( + timeoutMs = REGISTRATION_TIMEOUT_MS, +): Promise { + if (typeof navigator === 'undefined' || !('serviceWorker' in navigator)) return undefined; + + const existing = await navigator.serviceWorker.getRegistration('/'); + if (existing) return existing; + + return new Promise((resolve) => { + const timer = setTimeout(() => resolve(undefined), timeoutMs); + navigator.serviceWorker.ready.then( + (registration) => { + clearTimeout(timer); + resolve(registration); + }, + () => { + clearTimeout(timer); + resolve(undefined); + }, + ); + }); +} diff --git a/src/sw/app-sw.ts b/src/sw/app-sw.ts index 913db24..30df302 100644 --- a/src/sw/app-sw.ts +++ b/src/sw/app-sw.ts @@ -7,11 +7,8 @@ import { NetworkFirst, CacheFirst } from 'workbox-strategies'; import { ExpirationPlugin } from 'workbox-expiration'; import { CacheableResponsePlugin } from 'workbox-cacheable-response'; import { CACHE_PREFIX, CACHE_VERSION_SUFFIX, cacheName, purgeObsoleteCaches } from '../lib/swCache'; -import { - retentionErrorFromRpcMessage, - retentionGapFromError, - type RetentionGap, -} from '../lib/stellar/scannerCursor'; +import { fetchEventPages } from '../lib/stellar/rpcEvents'; +import { retentionGapFromError, type RetentionGap } from '../lib/stellar/scannerCursor'; import { deliverPushNotification, type NotificationPresentation, @@ -179,44 +176,13 @@ async function updateLastScannedLedger( } // ── Stellar RPC helpers ──────────────────────────────────────────────────────── - -async function fetchLatestLedger(): Promise { - const response = await fetch(STELLAR_RPC_URL, { - method: 'POST', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ jsonrpc: '2.0', id: 1, method: 'getLatestLedger' }), - }); - const data = await response.json(); - return data.result?.sequence || 0; -} - -async function fetchAnnouncementEvents( - startLedger: number, - contractId: string = ANNOUNCER_CONTRACT, -): Promise<{ events: unknown[]; latestLedger: number }> { - const response = await fetch(STELLAR_RPC_URL, { - method: 'POST', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ - jsonrpc: '2.0', - id: 2, - method: 'getEvents', - params: { - startLedger, - filters: [{ type: 'contract', contractIds: [contractId] }], - pagination: { limit: 1000 }, - }, - }), - }); - const data = await response.json(); - if (data.error?.message) { - const retentionError = retentionErrorFromRpcMessage(startLedger, String(data.error.message)); - throw retentionError ?? new Error(String(data.error.message)); - } - const events = data.result?.events || []; - const latestLedger = await fetchLatestLedger(); - return { events, latestLedger }; -} +// +// Event reads go through the shared, paginated implementation in +// src/lib/stellar/rpcEvents.ts. It follows the RPC cursor to the end of the +// committed boundary, maps out-of-range requests to retention gaps, and reports +// progress per page so a sync that is killed mid-way restarts at the first +// ledger proven fully announced instead of at the tip. A page boundary can +// split one ledger, so the boundary ledger is re-read rather than skipped. async function notifyRetentionGap(publicKey: string, gap: RetentionGap): Promise { const clients = await self.clients.matchAll({ type: 'window', includeUncontrolled: true }); @@ -228,14 +194,23 @@ async function notifyRetentionGap(publicKey: string, gap: RetentionGap): Promise async function scanStoredKey( db: IDBDatabase, storedKey: StoredViewingKey, - startLedger: number, + startLedger: number | undefined, ): Promise { - const { events, latestLedger } = await fetchAnnouncementEvents(startLedger); + const { events } = await fetchEventPages({ + rpcUrl: STELLAR_RPC_URL, + contractId: ANNOUNCER_CONTRACT, + // `undefined` lets the module clamp a first scan to the retention window + // instead of asking for a ledger the RPC has already dropped. + startLedger, + onPage: async (page) => { + await updateLastScannedLedger(db, storedKey.publicKey, page.committedLedger + 1); + }, + }); + if (events.length > 0) { - console.log(`Found ${events.length} events for ${storedKey.publicKey}`); + console.log(`[app-sw] Found ${events.length} events for ${storedKey.publicKey}`); // TODO: decrypt and scan with Wraith SDK when bundled in SW context } - await updateLastScannedLedger(db, storedKey.publicKey, latestLedger + 1); } // ── Background sync handler ──────────────────────────────────────────────────── @@ -252,17 +227,14 @@ async function handleSync(): Promise { }); for (const storedKey of allKeys) { - const startLedger = storedKey.lastScannedLedger || 1; try { - await scanStoredKey(db, storedKey, startLedger); + await scanStoredKey(db, storedKey, storedKey.lastScannedLedger); } catch (error) { const gap = retentionGapFromError(error); if (!gap) throw error; - if (storedKey.lastScannedLedger === undefined) { - await scanStoredKey(db, storedKey, gap.oldestAvailableLedger); - } else { - await notifyRetentionGap(storedKey.publicKey, gap); - } + // A stored cursor that fell out of the window is never advanced + // silently: surface the gap so the page can offer a recovery scan. + await notifyRetentionGap(storedKey.publicKey, gap); } } diff --git a/src/sw/stellar-notification-sw.ts b/src/sw/stellar-notification-sw.ts deleted file mode 100644 index d6d65c4..0000000 --- a/src/sw/stellar-notification-sw.ts +++ /dev/null @@ -1,563 +0,0 @@ -/** - * stellar-notification-sw.ts - * - * Stellar stealth-payment push notification service worker. - * - * Responsibilities: - * 1. Receive push events and show browser notifications. - * 2. Persist each notification into the app's zustand store by posting a - * message to all controlled clients so the React app can call - * `addNotification` on the next load (or immediately if a tab is open). - * 3. Periodic background sync to scan for new stealth payments. - * 4. IndexedDB storage for encrypted viewing keys. - * 5. Handle REGISTER_VIEWING_KEY / UNREGISTER_VIEWING_KEY messages from the - * client so background scanning knows which keys to watch. - * - * Push payload (JSON): - * { - * version: 1, - * id?: string, // otherwise derived from the validated payload - * title: string, - * body: string, - * amount?: string, // e.g. "12.5" - * asset?: string, // e.g. "XLM" - * sender?: string, // stealth / ephemeral address - * url?: string // same-origin path only - * } - */ - -/// -import { - retentionErrorFromRpcMessage, - retentionGapFromError, - type RetentionGap, -} from '../lib/stellar/scannerCursor'; -import { - deliverPushNotification, - type NotificationPresentation, - type ValidatedPushPayload, -} from './notificationPayload'; -import { - claimNotificationId, - createNotificationStore, - pruneNotificationIds, - releaseNotificationId, -} from './notificationStore'; -export {}; - -declare const self: ServiceWorkerGlobalScope; - -// ─── constants ──────────────────────────────────────────────────────────────── - -const NOTIFICATION_CHANNEL = 'wraith-notifications'; -const ANNOUNCER_CONTRACT = 'CCJLJ2QRBJAAKIG6ELNQVXLLWMKKWVN5O2FKWUETHZGMPAD4MHK7WVWL'; -const STELLAR_RPC_URL = 'https://soroban-testnet.stellar.org'; -const DB_NAME = 'wraith-stellar-notifications'; -const DB_VERSION = 2; -const STORE_NAME = 'viewing-keys'; -const SYNC_TAG = 'stellar-payment-scan'; -const SYNC_INTERVAL_MINUTES = 15; - -// ─── types ──────────────────────────────────────────────────────────────────── - -interface StoredViewingKey { - publicKey: string; - encryptedViewingKey: string; - encryptedSpendingPubKey: string; - encryptedSpendingScalar: string; - lastScannedLedger?: number; - timestamp: number; - relayUrl?: string; - metaAddressHash?: string; - pushSubscription?: PushSubscriptionJSON; -} - -interface PushSubscriptionJSON { - endpoint: string; - keys: { - p256dh: string; - auth: string; - }; -} - -interface NotificationData { - id: string; - stealthAddress?: string; - amount?: string; - asset?: string; - sender?: string; - timestamp: number; - url: string; -} - -// ─── IndexedDB helpers ──────────────────────────────────────────────────────── - -function openDB(): Promise { - return new Promise((resolve, reject) => { - const request = indexedDB.open(DB_NAME, DB_VERSION); - request.onerror = () => reject(request.error); - request.onsuccess = () => resolve(request.result); - request.onupgradeneeded = (event) => { - const db = (event.target as IDBOpenDBRequest).result; - if (!db.objectStoreNames.contains(STORE_NAME)) { - const store = db.createObjectStore(STORE_NAME, { keyPath: 'publicKey' }); - store.createIndex('timestamp', 'timestamp', { unique: false }); - } - createNotificationStore(db); - }; - }); -} - -async function updateLastScannedLedger( - db: IDBDatabase, - publicKey: string, - ledger: number, -): Promise { - return new Promise((resolve, reject) => { - const transaction = db.transaction([STORE_NAME], 'readwrite'); - const store = transaction.objectStore(STORE_NAME); - const request = store.get(publicKey); - request.onerror = () => reject(request.error); - request.onsuccess = () => { - const data = request.result as StoredViewingKey; - if (data) { - data.lastScannedLedger = ledger; - data.timestamp = Date.now(); - const updateRequest = store.put(data); - updateRequest.onerror = () => reject(updateRequest.error); - updateRequest.onsuccess = () => resolve(); - } else { - resolve(); - } - }; - }); -} - -// ─── Stellar RPC helpers ────────────────────────────────────────────────────── - -async function fetchLatestLedger(): Promise { - const response = await fetch(STELLAR_RPC_URL, { - method: 'POST', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ jsonrpc: '2.0', id: 1, method: 'getLatestLedger' }), - }); - const data = await response.json(); - return data.result?.sequence || 0; -} - -async function fetchAnnouncementEvents( - startLedger: number, - contractId: string = ANNOUNCER_CONTRACT, -): Promise<{ events: unknown[]; latestLedger: number }> { - const response = await fetch(STELLAR_RPC_URL, { - method: 'POST', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ - jsonrpc: '2.0', - id: 2, - method: 'getEvents', - params: { - startLedger, - filters: [{ type: 'contract', contractIds: [contractId] }], - pagination: { limit: 1000 }, - }, - }), - }); - const data = await response.json(); - if (data.error?.message) { - const retentionError = retentionErrorFromRpcMessage(startLedger, String(data.error.message)); - throw retentionError ?? new Error(String(data.error.message)); - } - const events = data.result?.events || []; - const latestLedger = await fetchLatestLedger(); - return { events, latestLedger }; -} - -async function notifyRetentionGap(publicKey: string, gap: RetentionGap): Promise { - const clients = await self.clients.matchAll({ type: 'window', includeUncontrolled: true }); - for (const client of clients) { - client.postMessage({ type: 'STELLAR_SCAN_RETENTION_GAP', publicKey, ...gap }); - } -} - -async function scanStoredKey( - db: IDBDatabase, - storedKey: StoredViewingKey, - startLedger: number, -): Promise { - const { events, latestLedger } = await fetchAnnouncementEvents(startLedger); - if (events.length > 0) { - console.log(`[wraith-sw] Found ${events.length} events for ${storedKey.publicKey}`); - // TODO: decrypt viewing key and scan with the SDK when bundled in the worker. - } - await updateLastScannedLedger(db, storedKey.publicKey, latestLedger + 1); -} - -// ─── push payload helpers ───────────────────────────────────────────────────── - -/** Broadcast to every open tab so the React store gets persisted immediately. */ -async function broadcastToClients(payload: ValidatedPushPayload, timestamp: number): Promise { - const clients = await self.clients.matchAll({ type: 'window', includeUncontrolled: true }); - for (const client of clients) { - client.postMessage({ - type: 'WRAITH_NOTIFICATION', - channel: NOTIFICATION_CHANNEL, - payload: { - ...payload, - timestamp, - }, - }); - } -} - -// ─── background sync ────────────────────────────────────────────────────────── - -async function handleSync(_event: ExtendableEvent): Promise { - try { - const db = await openDB(); - const allKeys = await new Promise((resolve, reject) => { - const transaction = db.transaction([STORE_NAME], 'readonly'); - const store = transaction.objectStore(STORE_NAME); - const request = store.getAll(); - request.onerror = () => reject(request.error); - request.onsuccess = () => resolve(request.result || []); - }); - - for (const storedKey of allKeys) { - const startLedger = storedKey.lastScannedLedger || 1; - try { - await scanStoredKey(db, storedKey, startLedger); - } catch (error) { - const gap = retentionGapFromError(error); - if (!gap) throw error; - if (storedKey.lastScannedLedger === undefined) { - await scanStoredKey(db, storedKey, gap.oldestAvailableLedger); - } else { - await notifyRetentionGap(storedKey.publicKey, gap); - } - } - } - - db.close(); - } catch (error) { - console.error('[wraith-sw] Background sync error:', error); - } -} - -// ─── push event ─────────────────────────────────────────────────────────────── - -self.addEventListener('push', (event: PushEvent) => { - event.waitUntil( - (async () => { - const db = await openDB(); - try { - const result = await deliverPushNotification( - event.data?.text() ?? '', - self.location.origin, - { - claim: (id) => claimNotificationId(db, id), - release: (id) => releaseNotificationId(db, id), - show: (presentation: NotificationPresentation) => - self.registration.showNotification(presentation.title, { - body: presentation.body, - icon: '/favicon-32x32.png', - badge: '/favicon-16x16.png', - tag: presentation.tag, - data: presentation.data, - }), - broadcast: broadcastToClients, - }, - ); - if (result === 'rejected') console.warn('[wraith-sw] Rejected invalid push payload'); - if (result === 'duplicate') console.log('[wraith-sw] Skipped duplicate notification'); - if (result === 'delivered') await pruneNotificationIds(db); - } finally { - db.close(); - } - })(), - ); -}); - -// ─── background sync event ──────────────────────────────────────────────────── - -self.addEventListener('sync', (event: SyncEvent) => { - if (event.tag === SYNC_TAG) { - event.waitUntil(handleSync(event)); - } -}); - -// ─── notification click ─────────────────────────────────────────────────────── - -self.addEventListener('notificationclick', (event: NotificationEvent) => { - const data = event.notification.data as NotificationData | undefined; - event.notification.close(); - - event.waitUntil( - self.clients.matchAll({ type: 'window', includeUncontrolled: true }).then((clientList) => { - // Focus existing Wraith tab and navigate to /notifications - const existing = clientList.find((c) => c.url.includes(self.location.origin) && 'focus' in c); - if (existing) { - (existing as WindowClient).focus(); - (existing as WindowClient).navigate(data?.url ?? '/notifications'); - // Also post match info so the page can pre-highlight it - if (data?.stealthAddress) { - existing.postMessage({ type: 'NAVIGATE_TO_MATCH', stealthAddress: data.stealthAddress }); - } - return; - } - return self.clients.openWindow(data?.url ?? '/notifications'); - }), - ); -}); - -// ─── message handler ────────────────────────────────────────────────────────── - -self.addEventListener('message', (event: ExtendableMessageEvent) => { - const { type, publicKey, encryptedViewingKey, encryptedSpendingPubKey, encryptedSpendingScalar } = - event.data ?? {}; - - if (type === 'RECOVER_SCAN_CURSOR') { - event.waitUntil( - (async () => { - const db = await openDB(); - try { - const recoveryLedger = Number(event.data.oldestAvailableLedger); - if ( - typeof publicKey !== 'string' || - !Number.isSafeInteger(recoveryLedger) || - recoveryLedger <= 0 - ) { - return; - } - const storedKey = await new Promise((resolve, reject) => { - const request = db - .transaction(STORE_NAME, 'readonly') - .objectStore(STORE_NAME) - .get(publicKey); - request.onerror = () => reject(request.error); - request.onsuccess = () => resolve(request.result); - }); - if (!storedKey) return; - await scanStoredKey(db, storedKey, recoveryLedger); - const clients = await self.clients.matchAll({ - type: 'window', - includeUncontrolled: true, - }); - clients.forEach((client) => - client.postMessage({ type: 'STELLAR_SCAN_RECOVERY_COMPLETE', publicKey }), - ); - } finally { - db.close(); - } - })(), - ); - return; - } - - if (type === 'REGISTER_VIEWING_KEY') { - event.waitUntil( - (async () => { - try { - const db = await openDB(); - const transaction = db.transaction([STORE_NAME], 'readwrite'); - const store = transaction.objectStore(STORE_NAME); - const entry: StoredViewingKey = { - publicKey, - encryptedViewingKey, - encryptedSpendingPubKey, - encryptedSpendingScalar, - timestamp: Date.now(), - }; - await new Promise((resolve, reject) => { - const request = store.put(entry); - request.onerror = () => reject(request.error); - request.onsuccess = () => resolve(); - }); - db.close(); - (event.source as Client)?.postMessage({ type: 'VIEWING_KEY_REGISTERED' }); - } catch (error) { - console.error('[wraith-sw] Failed to register viewing key:', error); - (event.source as Client)?.postMessage({ - type: 'VIEWING_KEY_ERROR', - error: error instanceof Error ? error.message : 'Unknown error', - }); - } - })(), - ); - } - - if (type === 'UNREGISTER_VIEWING_KEY') { - event.waitUntil( - (async () => { - try { - const db = await openDB(); - const transaction = db.transaction([STORE_NAME], 'readwrite'); - const store = transaction.objectStore(STORE_NAME); - await new Promise((resolve, reject) => { - const request = store.delete(publicKey); - request.onerror = () => reject(request.error); - request.onsuccess = () => resolve(); - }); - db.close(); - - // Unregister periodic sync if no keys remain - const db2 = await openDB(); - const remaining = await new Promise((resolve, reject) => { - const tx = db2.transaction([STORE_NAME], 'readonly'); - const st = tx.objectStore(STORE_NAME); - const req = st.getAll(); - req.onerror = () => reject(req.error); - req.onsuccess = () => resolve(req.result || []); - }); - db2.close(); - - if (remaining.length === 0 && 'periodicSync' in self.registration) { - await ( - self.registration as unknown as { - periodicSync: { unregister: (tag: string) => Promise }; - } - ).periodicSync.unregister(SYNC_TAG); - } - - (event.source as Client)?.postMessage({ type: 'VIEWING_KEY_UNREGISTERED' }); - } catch (error) { - console.error('[wraith-sw] Failed to unregister viewing key:', error); - } - })(), - ); - } - - if (type === 'TRIGGER_SCAN') { - event.waitUntil(handleSync(event as unknown as ExtendableEvent)); - } - - // Push subscription management - if (type === 'REGISTER_PUSH_SUBSCRIPTION') { - event.waitUntil( - (async () => { - try { - const { subscription, metaAddressHash, relayUrl } = event.data ?? {}; - if (!subscription || !metaAddressHash) { - throw new Error('Missing subscription or metaAddressHash'); - } - - // Store subscription info in IndexedDB - const db = await openDB(); - const transaction = db.transaction([STORE_NAME], 'readwrite'); - const store = transaction.objectStore(STORE_NAME); - - // Update existing entry or create new one - const existing = await new Promise((resolve, reject) => { - const request = store.get(subscription.keys?.p256dh || 'default'); - request.onerror = () => reject(request.error); - request.onsuccess = () => resolve(request.result); - }); - - const entry: StoredViewingKey = existing || { - publicKey: subscription.keys?.p256dh || 'default', - encryptedViewingKey: '', - encryptedSpendingPubKey: '', - encryptedSpendingScalar: '', - timestamp: Date.now(), - }; - - // Store relay URL and meta-address hash - (entry as any).relayUrl = relayUrl; - (entry as any).metaAddressHash = metaAddressHash; - (entry as any).pushSubscription = subscription; - - await new Promise((resolve, reject) => { - const request = store.put(entry); - request.onerror = () => reject(request.error); - request.onsuccess = () => resolve(); - }); - - db.close(); - (event.source as Client)?.postMessage({ type: 'PUSH_SUBSCRIPTION_REGISTERED' }); - } catch (error) { - console.error('[wraith-sw] Failed to register push subscription:', error); - (event.source as Client)?.postMessage({ - type: 'PUSH_SUBSCRIPTION_ERROR', - error: error instanceof Error ? error.message : 'Unknown error', - }); - } - })(), - ); - } - - if (type === 'UNREGISTER_PUSH_SUBSCRIPTION') { - event.waitUntil( - (async () => { - try { - const { subscription, metaAddressHash } = event.data ?? {}; - if (!subscription) { - throw new Error('Missing subscription'); - } - - const db = await openDB(); - const transaction = db.transaction([STORE_NAME], 'readwrite'); - const store = transaction.objectStore(STORE_NAME); - - // Remove push subscription from entry - const existing = await new Promise((resolve, reject) => { - const request = store.get(subscription.keys?.p256dh || 'default'); - request.onerror = () => reject(request.error); - request.onsuccess = () => resolve(request.result); - }); - - if (existing) { - delete (existing as any).relayUrl; - delete (existing as any).metaAddressHash; - delete (existing as any).pushSubscription; - - await new Promise((resolve, reject) => { - const request = store.put(existing); - request.onerror = () => reject(request.error); - request.onsuccess = () => resolve(); - }); - } - - db.close(); - (event.source as Client)?.postMessage({ type: 'PUSH_SUBSCRIPTION_UNREGISTERED' }); - } catch (error) { - console.error('[wraith-sw] Failed to unregister push subscription:', error); - (event.source as Client)?.postMessage({ - type: 'PUSH_SUBSCRIPTION_ERROR', - error: error instanceof Error ? error.message : 'Unknown error', - }); - } - })(), - ); - } -}); - -// ─── install / activate ─────────────────────────────────────────────────────── - -self.addEventListener('install', () => { - self.skipWaiting(); -}); - -self.addEventListener('activate', (event: ExtendableEvent) => { - event.waitUntil( - Promise.all([ - self.clients.claim(), - (async () => { - if ('periodicSync' in self.registration) { - try { - await ( - self.registration as unknown as { - periodicSync: { - register: (tag: string, opts: { minInterval: number }) => Promise; - }; - } - ).periodicSync.register(SYNC_TAG, { - minInterval: SYNC_INTERVAL_MINUTES * 60 * 1000, - }); - } catch { - // periodicSync not supported in this environment — silently skip - } - } - })(), - ]), - ); -}); diff --git a/src/workers/stellar-scanner.worker.test.ts b/src/workers/stellar-scanner.worker.test.ts index 35c4dec..30cdc5e 100644 --- a/src/workers/stellar-scanner.worker.test.ts +++ b/src/workers/stellar-scanner.worker.test.ts @@ -1,4 +1,4 @@ -import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { afterEach, describe, expect, it, vi } from 'vitest'; import { Address, nativeToScVal, xdr } from '@stellar/stellar-sdk'; import { deriveStealthKeys, generateStealthAddress } from '@wraith-protocol/sdk/chains/stellar'; @@ -12,6 +12,11 @@ import { deriveStealthKeys, generateStealthAddress } from '@wraith-protocol/sdk/ const RPC_URL = 'https://soroban-testnet.stellar.org'; const CONTRACT_ID = 'CCJLJ2QRBJAAKIG6ELNQVXLLWMKKWVN5O2FKWUETHZGMPAD4MHK7WVWL'; +interface WorkerMessage { + type: string; + [key: string]: unknown; +} + function randomSignature(): Uint8Array { const bytes = new Uint8Array(64); crypto.getRandomValues(bytes); @@ -46,20 +51,71 @@ function buildEventEntry(params: { }; } +function jsonResponse(payload: unknown): Response { + return new Response(JSON.stringify({ jsonrpc: '2.0', id: 1, ...(payload as object) })); +} + function mockGetEventsResponse(events: ReturnType[]) { vi.stubGlobal( 'fetch', vi.fn(async (_url: string, init?: RequestInit) => { const body = JSON.parse((init?.body as string) ?? '{}'); if (body.method === 'getLatestLedger') { - return new Response( - JSON.stringify({ jsonrpc: '2.0', id: body.id, result: { sequence: 500 } }), - ); + return jsonResponse({ result: { sequence: 500 } }); } if (body.method !== 'getEvents') { - return new Response(JSON.stringify({ jsonrpc: '2.0', id: body.id, result: {} })); + return jsonResponse({ result: {} }); } - return new Response(JSON.stringify({ jsonrpc: '2.0', id: body.id, result: { events } })); + return jsonResponse({ result: { events } }); + }), + ); +} + +/** + * Serves `getEvents` the way Soroban RPC does: a probe request (limit 1), then + * one response per page, each carrying the cursor for the next page. + */ +function mockPagedGetEvents(options: { + pages: ReturnType[][]; + latestLedger?: number; + /** 1-based data-page index that should fail with the given status. */ + failPage?: { page: number; status: number }; + requests?: Record[]; +}) { + const latestLedger = options.latestLedger ?? 5_000; + let dataPage = 0; + + vi.stubGlobal( + 'fetch', + vi.fn(async (_url: string, init?: RequestInit) => { + const body = JSON.parse((init?.body as string) ?? '{}'); + options.requests?.push(body); + + if (body.method === 'getLatestLedger') { + return jsonResponse({ result: { sequence: latestLedger } }); + } + if (body.method !== 'getEvents') { + return jsonResponse({ result: {} }); + } + if (body.params?.pagination?.limit === 1) { + // Retention probe: the whole history is inside the window. + return jsonResponse({ result: { events: [], latestLedger } }); + } + + dataPage += 1; + if (options.failPage && options.failPage.page === dataPage) { + return new Response('upstream unavailable', { status: options.failPage.status }); + } + + const events = options.pages[dataPage - 1] ?? []; + const hasMore = dataPage < options.pages.length; + return jsonResponse({ + result: { + events, + latestLedger, + ...(hasMore ? { cursor: `cursor-${dataPage}` } : {}), + }, + }); }), ); } @@ -76,18 +132,45 @@ describe('stellar-scanner worker message contract', () => { vi.resetModules(); }); - async function postAndWaitForResult(message: Record) { + /** + * Posts a scan request and collects every message until the worker reports a + * terminal one (SUCCESS / ERROR / RETENTION_GAP). PAGE messages arrive first. + */ + async function runScan( + message: Record, + ): Promise<{ messages: WorkerMessage[]; terminal: WorkerMessage }> { const worker = (globalThis as unknown as { self: any }).self; - return new Promise((resolve) => { + return new Promise((resolve) => { + const messages: WorkerMessage[] = []; const originalPostMessage = worker.postMessage; - worker.postMessage = (data: unknown) => { - worker.postMessage = originalPostMessage; - resolve(data); + worker.postMessage = (data: WorkerMessage) => { + messages.push(data); + if (['SUCCESS', 'ERROR', 'RETENTION_GAP'].includes(data?.type)) { + worker.postMessage = originalPostMessage; + resolve({ messages, terminal: data }); + } }; worker.onmessage({ data: message }); }); } + async function postAndWaitForResult(message: Record) { + const { terminal } = await runScan(message); + return terminal; + } + + function validEvent() { + const other = deriveStealthKeys(randomSignature()); + const generated = generateStealthAddress(other.spendingPubKey, other.viewingPubKey); + return buildEventEntry({ + schemeId: 1, + stealthAddress: generated.stealthAddress, + caller, + ephemeralPubKey: generated.ephemeralPubKey, + viewTag: generated.viewTag, + }); + } + it('fast strategy: dispatches through the real onmessage handler and finds a genuine payment', async () => { await import('./stellar-scanner.worker'); @@ -113,7 +196,7 @@ describe('stellar-scanner worker message contract', () => { expect(result.type).toBe('SUCCESS'); expect(result.results).toHaveLength(1); - expect(result.results[0].stealthAddress).toBe(generated.stealthAddress); + expect((result.results as any)[0].stealthAddress).toBe(generated.stealthAddress); }); it('balanced strategy (default, no strategy field) matches the SDK default behavior', async () => { @@ -188,7 +271,7 @@ describe('stellar-scanner worker message contract', () => { expect(fullResult.type).toBe('SUCCESS'); expect(fullResult.results).toHaveLength(1); - expect(fullResult.results[0].stealthAddress).toBe(generated.stealthAddress); + expect((fullResult.results as any)[0].stealthAddress).toBe(generated.stealthAddress); }); it('posts an ERROR message if the RPC call throws unexpectedly', async () => { @@ -222,19 +305,17 @@ describe('stellar-scanner worker message contract', () => { vi.fn(async (_url: string, init?: RequestInit) => { const body = JSON.parse((init?.body as string) ?? '{}'); if (body.method === 'getLatestLedger') { - return new Response(JSON.stringify({ result: { sequence: 500 } })); + return jsonResponse({ result: { sequence: 500 } }); } const startLedger = body.params?.startLedger; if (startLedger === 1) { - return new Response( - JSON.stringify({ - error: { message: 'startLedger must be within the ledger range: 250 - 500' }, - }), - ); + return jsonResponse({ + error: { message: 'startLedger must be within the ledger range: 250 - 500' }, + }); } requestedStartLedgers.push(startLedger); - return new Response(JSON.stringify({ result: { events: [] } })); + return jsonResponse({ result: { events: [] } }); }), ); @@ -256,7 +337,126 @@ describe('stellar-scanner worker message contract', () => { expect(requestedStartLedgers).toEqual([]); const recovered = await postAndWaitForResult({ ...message, startLedger: 250 }); - expect(recovered).toEqual({ type: 'SUCCESS', results: [], nextLedger: 501 }); + expect(recovered).toMatchObject({ type: 'SUCCESS', results: [], nextLedger: 501 }); expect(requestedStartLedgers).toEqual([250]); }); + + it('paginates past the 1,000-event page cap and reports progress for every page', async () => { + await import('./stellar-scanner.worker'); + const requests: Record[] = []; + const generated = generateStealthAddress(recipient.spendingPubKey, recipient.viewingPubKey); + const filler = validEvent(); + + mockPagedGetEvents({ + latestLedger: 5_000, + requests, + pages: [ + Array.from({ length: 1_000 }, () => ({ ...filler, ledger: 100 })), + [ + { + ...buildEventEntry({ + schemeId: 1, + stealthAddress: generated.stealthAddress, + caller, + ephemeralPubKey: generated.ephemeralPubKey, + viewTag: generated.viewTag, + }), + ledger: 120, + }, + ], + ], + }); + + const { messages, terminal } = await runScan({ + rpcUrl: RPC_URL, + announcerContract: CONTRACT_ID, + viewingKey: recipient.viewingKey, + spendingPubKey: recipient.spendingPubKey, + spendingScalar: recipient.spendingScalar, + strategy: 'balanced', + }); + + const pages = messages.filter((message) => message.type === 'PAGE'); + expect(pages).toHaveLength(2); + // All 1,000 events share ledger 100 and the page still has a cursor, so the + // ledger could continue on the next page and may not be committed yet. + expect(pages[0]).toMatchObject({ page: 1, events: 1_000, committedLedger: 99 }); + // The second page is only reachable by following the RPC cursor. + const dataRequests = requests.filter( + (request) => request.method === 'getEvents' && request.params?.pagination?.limit !== 1, + ); + expect(dataRequests[0].params.startLedger).toBe(1); + expect(dataRequests[1].params.pagination.cursor).toBe('cursor-1'); + + expect(terminal.type).toBe('SUCCESS'); + expect(terminal.results).toHaveLength(1); + expect((terminal.results as any)[0].stealthAddress).toBe(generated.stealthAddress); + }); + + it('keeps the committed cursor when a later page fails so the rescan resumes there', async () => { + await import('./stellar-scanner.worker'); + const requests: Record[] = []; + const generated = generateStealthAddress(recipient.spendingPubKey, recipient.viewingPubKey); + + const pages = [ + Array.from({ length: 3 }, () => ({ ...validEvent(), ledger: 1_100 })), + [ + { + ...buildEventEntry({ + schemeId: 1, + stealthAddress: generated.stealthAddress, + caller, + ephemeralPubKey: generated.ephemeralPubKey, + viewTag: generated.viewTag, + }), + ledger: 1_200, + }, + ], + ]; + + mockPagedGetEvents({ + latestLedger: 5_000, + requests, + pages, + failPage: { page: 2, status: 503 }, + }); + const { messages: failedMessages, terminal: failed } = await runScan({ + rpcUrl: RPC_URL, + announcerContract: CONTRACT_ID, + viewingKey: recipient.viewingKey, + spendingPubKey: recipient.spendingPubKey, + spendingScalar: recipient.spendingScalar, + strategy: 'balanced', + startLedger: 1_000, + }); + + expect(failed.type).toBe('ERROR'); + + // The first page stopped inside ledger 1,100, so its remaining events stay + // uncommitted and the persisted cursor points back at that ledger. + const firstPage = failedMessages.find((message) => message.type === 'PAGE'); + expect(firstPage).toMatchObject({ page: 1, committedLedger: 1_099 }); + const persistedStartLedger = Number(firstPage?.committedLedger) + 1; + expect(persistedStartLedger).toBe(1_100); + + // Rescan from the cursor the page persisted: the failed page must still be + // read, and nothing after it may be skipped. + mockPagedGetEvents({ latestLedger: 5_000, requests, pages: [pages[1]] }); + const resumed = await postAndWaitForResult({ + rpcUrl: RPC_URL, + announcerContract: CONTRACT_ID, + viewingKey: recipient.viewingKey, + spendingPubKey: recipient.spendingPubKey, + spendingScalar: recipient.spendingScalar, + strategy: 'balanced', + startLedger: persistedStartLedger, + }); + + const resumedDataRequests = requests.filter( + (request) => request.method === 'getEvents' && request.params?.pagination?.limit !== 1, + ); + expect(resumedDataRequests[resumedDataRequests.length - 1].params.startLedger).toBe(1_100); + expect(resumed.type).toBe('SUCCESS'); + expect(resumed.results).toHaveLength(1); + }); }); diff --git a/src/workers/stellar-scanner.worker.ts b/src/workers/stellar-scanner.worker.ts index bf86b41..4bca13e 100644 --- a/src/workers/stellar-scanner.worker.ts +++ b/src/workers/stellar-scanner.worker.ts @@ -1,147 +1,49 @@ -import { bytesToHex } from '@wraith-protocol/sdk/chains/stellar'; -import type { Announcement } from '@wraith-protocol/sdk/chains/stellar'; -import { Address, xdr } from '@stellar/stellar-sdk'; +import type { MatchedAnnouncement } from '@wraith-protocol/sdk/chains/stellar'; import { scanWithStrategy, DEFAULT_SCAN_STRATEGY, type ScanStrategy } from './stellarScanDispatch'; -import { retentionErrorFromRpcMessage, retentionGapFromError } from '../lib/stellar/scannerCursor'; - -function parseLedgerRange(message: string): { oldest: number; latest: number } | undefined { - const match = message.match(/range:\s*(\d+)\s*-\s*(\d+)/i); - if (!match) return undefined; - return { oldest: Number(match[1]), latest: Number(match[2]) }; +import { retentionGapFromError } from '../lib/stellar/scannerCursor'; +import { fetchAnnouncementEvents } from '../lib/stellar/announcementRpc'; + +export interface StellarScannerRequest { + rpcUrl: string; + announcerContract: string; + viewingKey: Uint8Array; + spendingPubKey: Uint8Array; + spendingScalar: bigint; + strategy?: ScanStrategy; + startLedger?: number; + /** Page size override; the default is the RPC's 1,000-event cap. */ + pageLimit?: number; } -async function getLatestLedger(rpcUrl: string): Promise { - const response = await fetch(rpcUrl, { - method: 'POST', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ jsonrpc: '2.0', id: 1, method: 'getLatestLedger' }), - }); - const data = await response.json(); - const sequence = Number(data.result?.sequence); - if (!Number.isSafeInteger(sequence) || sequence <= 0) { - throw new Error('Stellar RPC did not return a valid latest ledger'); - } - return sequence; -} - -async function fetchAnnouncementEvents( - rpcUrl: string, - contractId: string, - savedStartLedger?: number, -): Promise<{ announcements: Announcement[]; nextLedger: number }> { - const all: Announcement[] = []; - 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(); - const ledgerWindow = probeData.error?.message - ? parseLedgerRange(String(probeData.error.message)) - : undefined; - const latestLedger = ledgerWindow?.latest ?? (await getLatestLedger(rpcUrl)); - const startLedger = - savedStartLedger ?? Math.max(ledgerWindow?.oldest ?? 1, Math.max(1, latestLedger - 5000)); - - if (ledgerWindow && startLedger < ledgerWindow.oldest) { - throw retentionErrorFromRpcMessage( - startLedger, - `ledger range: ${ledgerWindow.oldest} - ${ledgerWindow.latest}`, - ); - } - if (probeData.error?.message && !ledgerWindow) { - throw new Error(String(probeData.error.message)); - } - - let cursor: string | undefined; - let hasMore = true; - - while (hasMore) { - const params: Record = { - filters: [{ type: 'contract', contractIds: [contractId] }], - pagination: { limit: 1000 }, - }; - - if (cursor) { - (params.pagination as Record).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(); - if (data.error?.message) { - const range = parseLedgerRange(String(data.error.message)); - if (range && startLedger < range.oldest) { - throw retentionErrorFromRpcMessage(startLedger, String(data.error.message)); - } - throw new Error(String(data.error.message)); +/** + * Messages the worker posts back to the page. + * + * `PAGE` is emitted once per processed page, before that page's progress is + * persisted, so an interrupted scan resumes from the first ledger it has not + * delivered rather than from the tip. `SUCCESS` follows with the aggregate. + */ +export type StellarScannerMessage = + | { + type: 'PAGE'; + page: number; + committedLedger: number; + events: number; + undecodableEvents: number; + matched: MatchedAnnouncement[]; } - const events = data.result?.events ?? []; - - for (const event of events) { - try { - const ann = parseAnnouncementEvent(event); - if (ann) all.push(ann); - } catch { - // Skip malformed events without discarding the scan cursor. - } + | { + type: 'SUCCESS'; + results: MatchedAnnouncement[]; + nextLedger: number; + committedLedger: number; + pages: number; + truncated: boolean; } + | { type: 'RETENTION_GAP'; requestedLedger: number; oldestAvailableLedger: number } + | { type: 'ERROR'; error: string }; - if (events.length < 1000) { - hasMore = false; - } else { - cursor = data.result?.cursor; - if (!cursor) hasMore = false; - } - } - - return { announcements: all, nextLedger: latestLedger + 1 }; -} - -function parseAnnouncementEvent(event: Record): 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 postMessage(message: StellarScannerMessage): void { + self.postMessage(message); } self.onmessage = async (e: MessageEvent) => { @@ -153,37 +55,53 @@ self.onmessage = async (e: MessageEvent) => { spendingScalar, strategy, startLedger, - }: { - rpcUrl: string; - announcerContract: string; - viewingKey: Uint8Array; - spendingPubKey: Uint8Array; - spendingScalar: bigint; - strategy?: ScanStrategy; - startLedger?: number; - } = e.data; + pageLimit, + }: StellarScannerRequest = e.data; try { - const { announcements, nextLedger } = await fetchAnnouncementEvents( + const scanStrategy = strategy ?? DEFAULT_SCAN_STRATEGY; + const results: MatchedAnnouncement[] = []; + + const { nextLedger, committedLedger, pages, truncated } = await fetchAnnouncementEvents({ rpcUrl, - announcerContract, + contractId: announcerContract, startLedger, - ); - const results = scanWithStrategy( - strategy ?? DEFAULT_SCAN_STRATEGY, - announcements, - viewingKey, - spendingPubKey, - spendingScalar, - ); - self.postMessage({ type: 'SUCCESS', results, nextLedger }); + pageLimit, + onPage: (page) => { + const matched = scanWithStrategy( + scanStrategy, + page.announcements, + viewingKey, + spendingPubKey, + spendingScalar, + ); + results.push(...matched); + postMessage({ + type: 'PAGE', + page: page.page, + committedLedger: page.committedLedger, + events: page.announcements.length, + undecodableEvents: page.undecodableEvents, + matched, + }); + }, + }); + + postMessage({ + type: 'SUCCESS', + results, + nextLedger, + committedLedger, + pages, + truncated, + }); } catch (err) { const retentionGap = retentionGapFromError(err); if (retentionGap) { - self.postMessage({ type: 'RETENTION_GAP', ...retentionGap }); + postMessage({ type: 'RETENTION_GAP', ...retentionGap }); return; } - self.postMessage({ + postMessage({ type: 'ERROR', error: err instanceof Error ? err.message : 'Scan failed in worker', }); diff --git a/vite.config.ts b/vite.config.ts index 3195d92..d2383ff 100644 --- a/vite.config.ts +++ b/vite.config.ts @@ -83,8 +83,8 @@ export default defineConfig({ injectManifest: { // Precache all built assets (JS, CSS, HTML, icons, fonts) globPatterns: ['**/*.{js,css,html,ico,png,svg,woff2}'], - // Exclude the stellar-notification-sw (it's now merged into app-sw) - globIgnores: ['**/stellar-notification-sw*', '**/app-sw*'], + // app-sw.ts is the service worker itself, not a prefetchable asset + globIgnores: ['**/app-sw*'], // Raise file size limit — Stellar SDK + wagmi chunks exceed 2 MiB maximumFileSizeToCacheInBytes: 4 * 1024 * 1024, // 4 MiB },