diff --git a/src/lib/actions/watchAnnouncementsForUser/watchAnnouncementsForUser.test.ts b/src/lib/actions/watchAnnouncementsForUser/watchAnnouncementsForUser.test.ts index 3a9b6ef..1c885a1 100644 --- a/src/lib/actions/watchAnnouncementsForUser/watchAnnouncementsForUser.test.ts +++ b/src/lib/actions/watchAnnouncementsForUser/watchAnnouncementsForUser.test.ts @@ -1,5 +1,5 @@ import * as BunTest from 'bun:test'; -import { type Address, getAddress } from 'viem'; +import { type Address, type PublicClient, getAddress } from 'viem'; import { type AnnouncementLog, ERC5564AnnouncerAbi, @@ -12,7 +12,7 @@ import setupTestStealthKeys from '../../helpers/test/setupTestStealthKeys'; import setupTestWallet from '../../helpers/test/setupTestWallet'; import type { SuperWalletClient } from '../../helpers/types'; import type { StealthActions } from '../../stealthClient/types'; -import { +import watchAnnouncementsForUser, { createWatchedAnnouncementsQueue, processWatchedAnnouncementsBatch, startWatchHeartbeat @@ -110,6 +110,54 @@ const announce = async ({ }; }; +const getNextBlockNumber = async (walletClient: SuperWalletClient) => + (await walletClient.getBlockNumber({ cacheTime: 0 })) + 1n; + +describe('watchAnnouncementsForUser setup failures', () => { + test('does not start heartbeat when polling setup fails', async () => { + const setupError = new Error('initial block number failed'); + const reportedErrors: string[] = []; + let getBlockNumberCalls = 0; + const schemeId = VALID_SCHEME_ID.SCHEME_ID_1; + const schemeIdBigInt = BigInt(schemeId); + const { spendingPublicKey, viewingPrivateKey } = + setupTestStealthKeys(schemeId); + + await expect( + watchAnnouncementsForUser({ + ERC5564Address: ERC5564_CONTRACT_ADDRESS as Address, + args: { + schemeId: schemeIdBigInt, + caller: '0x00000000000000000000000000000000000000AA' + }, + clientParams: { + publicClient: { + getBlockNumber: async () => { + getBlockNumberCalls += 1; + throw setupError; + } + } as unknown as PublicClient + }, + handleLogsForUser: () => {}, + onError: error => { + reportedErrors.push(error.message); + }, + onHeartbeat: () => {}, + pollOptions: { + pollingInterval: 10 + }, + spendingPublicKey, + viewingPrivateKey + }) + ).rejects.toThrow('initial block number failed'); + + await sleep(35); + + expect(getBlockNumberCalls).toEqual(1); + expect(reportedErrors).toHaveLength(0); + }); +}); + describe('watchAnnouncementsForUser', () => { let stealthClient: StealthActions; let walletClient: SuperWalletClient; @@ -131,7 +179,7 @@ describe('watchAnnouncementsForUser', () => { test('awaits async handlers and still delivers relevant announcements', async () => { const watchedAnnouncements: AnnouncementLog[] = []; - const fromBlock = (await walletClient.getBlockNumber()) + 1n; + const fromBlock = await getNextBlockNumber(walletClient); const unwatch = await stealthClient.watchAnnouncementsForUser({ ERC5564Address, args: { @@ -186,7 +234,7 @@ describe('watchAnnouncementsForUser', () => { test('reports rejected handlers once and keeps watching later announcements', async () => { const handledAnnouncements: AnnouncementLog[] = []; const handlerErrors: string[] = []; - const fromBlock = (await walletClient.getBlockNumber()) + 1n; + const fromBlock = await getNextBlockNumber(walletClient); let handlerCalls = 0; const unwatch = await stealthClient.watchAnnouncementsForUser({ @@ -272,7 +320,7 @@ describe('watchAnnouncementsForUser', () => { const loggedErrors: Error[] = []; const handledAnnouncements: AnnouncementLog[] = []; let handlerCalls = 0; - const fromBlock = (await walletClient.getBlockNumber()) + 1n; + const fromBlock = await getNextBlockNumber(walletClient); console.error = ((value: unknown) => { if (value instanceof Error) { @@ -360,7 +408,7 @@ describe('watchAnnouncementsForUser', () => { const originalConsoleError = console.error; const loggedErrors: Error[] = []; const handledAnnouncements: AnnouncementLog[] = []; - const fromBlock = (await walletClient.getBlockNumber()) + 1n; + const fromBlock = await getNextBlockNumber(walletClient); let handlerCalls = 0; console.error = ((value: unknown) => { @@ -452,7 +500,7 @@ describe('watchAnnouncementsForUser', () => { test('recovers when fallback console logging throws', async () => { const originalConsoleError = console.error; const handledAnnouncements: AnnouncementLog[] = []; - const fromBlock = (await walletClient.getBlockNumber()) + 1n; + const fromBlock = await getNextBlockNumber(walletClient); let consoleErrorCalls = 0; let handlerCalls = 0; @@ -613,7 +661,7 @@ describe('watchAnnouncementsForUser', () => { rawLogCount: number; relevantLogCount: number; }> = []; - const fromBlock = (await walletClient.getBlockNumber()) + 1n; + const fromBlock = await getNextBlockNumber(walletClient); const unwatch = await stealthClient.watchAnnouncementsForUser({ ERC5564Address, @@ -697,7 +745,7 @@ describe('watchAnnouncementsForUser', () => { let activeHandlers = 0; let handlerCalls = 0; let overlapDetected = false; - const fromBlock = (await walletClient.getBlockNumber()) + 1n; + const fromBlock = await getNextBlockNumber(walletClient); const unwatch = await stealthClient.watchAnnouncementsForUser({ ERC5564Address, @@ -791,7 +839,7 @@ describe('watchAnnouncementsForUser', () => { const firstHandlerStarted = createDeferred(); const firstBatchFinished = createDeferred(); const handledAnnouncements: AnnouncementLog[] = []; - const fromBlock = (await walletClient.getBlockNumber()) + 1n; + const fromBlock = await getNextBlockNumber(walletClient); let sawLaterBatch = false; let sawFirstBatch = false; @@ -879,7 +927,7 @@ describe('watchAnnouncementsForUser', () => { test('does not emit announcements that do not apply to the user', async () => { const watchedAnnouncements: AnnouncementLog[] = []; - const fromBlock = (await walletClient.getBlockNumber()) + 1n; + const fromBlock = await getNextBlockNumber(walletClient); const unwatch = await stealthClient.watchAnnouncementsForUser({ ERC5564Address, args: { diff --git a/src/lib/actions/watchAnnouncementsForUser/watchAnnouncementsForUser.ts b/src/lib/actions/watchAnnouncementsForUser/watchAnnouncementsForUser.ts index 8ad1b52..4cc12eb 100644 --- a/src/lib/actions/watchAnnouncementsForUser/watchAnnouncementsForUser.ts +++ b/src/lib/actions/watchAnnouncementsForUser/watchAnnouncementsForUser.ts @@ -57,6 +57,7 @@ async function reportWatchError({ } const DEFAULT_WATCH_HEARTBEAT_INTERVAL_MS = 4_000; +const DEFAULT_WATCH_POLLING_INTERVAL_MS = 4_000; type WatchedAnnouncementsQueue = { enqueue: (task: () => Promise) => void; @@ -96,6 +97,20 @@ type WatchedAnnouncementEventLog = Omit< >; }; +const getInitialPreviousBlockNumber = async ({ + fromBlock, + publicClient +}: { + fromBlock?: WatchAnnouncementsForUserParams['fromBlock']; + publicClient: ReturnType; +}): Promise => { + if (typeof fromBlock === 'bigint') { + return fromBlock - 1n; + } + + return publicClient.getBlockNumber({ cacheTime: 0 }); +}; + const getObservedBlockFromLogs = ( logs: WatchedAnnouncementEventLog[] ): bigint | undefined => @@ -245,6 +260,136 @@ export function startWatchHeartbeat({ }; } +export async function startPollingAnnouncementEvents({ + args, + ERC5564Address, + excludeList, + fromBlock, + handleLogsForUser, + includeList, + onError, + pollOptions, + publicClient, + spendingPublicKey, + viewingPrivateKey, + watchedAnnouncementsQueue +}: { + args: WatchAnnouncementsForUserParams['args']; + ERC5564Address: WatchAnnouncementsForUserParams['ERC5564Address']; + excludeList?: WatchAnnouncementsForUserParams['excludeList']; + fromBlock?: WatchAnnouncementsForUserParams['fromBlock']; + handleLogsForUser: WatchAnnouncementsForUserParams['handleLogsForUser']; + includeList?: WatchAnnouncementsForUserParams['includeList']; + onError?: WatchAnnouncementsForUserParams['onError']; + pollOptions?: WatchAnnouncementsForUserParams['pollOptions']; + publicClient: ReturnType; + spendingPublicKey: WatchAnnouncementsForUserParams['spendingPublicKey']; + viewingPrivateKey: WatchAnnouncementsForUserParams['viewingPrivateKey']; + watchedAnnouncementsQueue: WatchedAnnouncementsQueue; +}): Promise { + let active = true; + let timeoutId: ReturnType | undefined; + let previousBlockNumber = await getInitialPreviousBlockNumber({ + fromBlock, + publicClient + }); + const pollingInterval = + pollOptions?.pollingInterval ?? + publicClient.pollingInterval ?? + DEFAULT_WATCH_POLLING_INTERVAL_MS; + const enqueueLogs = (logs: WatchedAnnouncementEventLog[]) => { + if (logs.length === 0) { + return; + } + + const batches = + pollOptions?.batch === false ? logs.map(log => [log]) : [logs]; + + for (const batch of batches) { + watchedAnnouncementsQueue.enqueue(async () => { + await processWatchedAnnouncementsBatch({ + logs: batch, + spendingPublicKey, + viewingPrivateKey, + publicClient, + excludeList, + includeList, + fromBlock, + handleLogsForUser, + onError + }); + }); + } + }; + + const pollAnnouncements = async () => { + try { + const blockNumber = await publicClient.getBlockNumber({ cacheTime: 0 }); + if (!active) { + return; + } + + const nextFromBlock = previousBlockNumber + 1n; + + if (nextFromBlock > blockNumber) { + return; + } + + const logs = (await publicClient.getContractEvents({ + address: ERC5564Address, + abi: ERC5564AnnouncerAbi, + eventName: 'Announcement', + args, + fromBlock: nextFromBlock, + toBlock: blockNumber, + strict: true + })) as WatchedAnnouncementEventLog[]; + + if (!active) { + return; + } + + previousBlockNumber = blockNumber; + enqueueLogs(logs); + } catch (error) { + await reportWatchError({ + error, + fallbackMessage: + 'watchAnnouncementsForUser failed while polling announcement events', + onError + }); + } + }; + + const scheduleNextPoll = () => { + timeoutId = setTimeout(() => { + void runPollLoop(); + }, pollingInterval); + }; + + const runPollLoop = async () => { + if (!active) { + return; + } + + await pollAnnouncements(); + + if (active) { + scheduleNextPoll(); + } + }; + + void runPollLoop(); + + return () => { + active = false; + + if (timeoutId !== undefined) { + clearTimeout(timeoutId); + } + }; +} + /** * Watches for announcement events relevant to the user. * @@ -285,6 +430,22 @@ async function watchAnnouncementsForUser({ }); } }); + // Poll with getContractEvents directly so logs cannot slip through while an RPC filter is still being created. + const unwatchContractEvent = await startPollingAnnouncementEvents({ + args, + ERC5564Address, + excludeList, + fromBlock, + handleLogsForUser, + includeList, + onError, + pollOptions, + publicClient, + spendingPublicKey, + viewingPrivateKey, + watchedAnnouncementsQueue + }); + const stopHeartbeat = startWatchHeartbeat({ fromBlock, onError, @@ -293,31 +454,6 @@ async function watchAnnouncementsForUser({ publicClient }); - const unwatchContractEvent = publicClient.watchContractEvent({ - address: ERC5564Address, - abi: ERC5564AnnouncerAbi, - eventName: 'Announcement', - args, - ...(fromBlock === 'latest' ? {} : { fromBlock }), - onLogs: logs => { - watchedAnnouncementsQueue.enqueue(async () => { - await processWatchedAnnouncementsBatch({ - logs, - spendingPublicKey, - viewingPrivateKey, - publicClient, - excludeList, - includeList, - fromBlock, - handleLogsForUser, - onError - }); - }); - }, - strict: true, - ...pollOptions - }); - return () => { stopHeartbeat(); unwatchContractEvent();