diff --git a/.github/screenshots/explorer-live-feed/before-highlight.jpg b/.github/screenshots/explorer-live-feed/before-highlight.jpg new file mode 100644 index 000000000..004738213 Binary files /dev/null and b/.github/screenshots/explorer-live-feed/before-highlight.jpg differ diff --git a/.github/screenshots/explorer-live-feed/live-reference.png b/.github/screenshots/explorer-live-feed/live-reference.png new file mode 100644 index 000000000..7a8f5f200 Binary files /dev/null and b/.github/screenshots/explorer-live-feed/live-reference.png differ diff --git a/.github/screenshots/explorer-live-feed/new-rows.jpg b/.github/screenshots/explorer-live-feed/new-rows.jpg new file mode 100644 index 000000000..7e9d83be5 Binary files /dev/null and b/.github/screenshots/explorer-live-feed/new-rows.jpg differ diff --git a/.github/screenshots/explorer-live-feed/paused-reference.png b/.github/screenshots/explorer-live-feed/paused-reference.png new file mode 100644 index 000000000..06581187c Binary files /dev/null and b/.github/screenshots/explorer-live-feed/paused-reference.png differ diff --git a/apps/explorer/src/comps/DataGrid.tsx b/apps/explorer/src/comps/DataGrid.tsx index bb7fe5ef9..61f4dea51 100644 --- a/apps/explorer/src/comps/DataGrid.tsx +++ b/apps/explorer/src/comps/DataGrid.tsx @@ -4,6 +4,7 @@ import * as React from 'react' import { Pagination } from '#comps/Pagination' import { Sections } from '#comps/Sections' import { cx } from '#lib/css' +import { useNewLiveRows } from '#lib/use-new-live-rows' import ChevronDownIcon from '~icons/lucide/chevron-down' export function DataGrid(props: DataGrid.Props) { @@ -56,6 +57,10 @@ export function DataGrid(props: DataGrid.Props) { }), })) : items(mode) + const newLiveRows = useNewLiveRows( + activeItems.map((item) => item.key), + effectiveLoading ? undefined : props.liveScope, + ) const pages = pagesProp ?? Math.ceil(totalItems / itemsPerPage) const isSimpleSinglePage = pagination === 'simple' && @@ -145,6 +150,9 @@ export function DataGrid(props: DataGrid.Props) { 'hover:bg-surface-hover transition-colors duration-100', item.expanded && 'border-l-distinct', item.className, + item.key !== undefined && + newLiveRows.has(item.key) && + 'block-row-shimmer', )} > {item.link && ( @@ -278,6 +286,8 @@ export namespace DataGrid { } export interface Props { + /** Enables new-row highlights, resetting when the feed scope changes. */ + liveScope?: string | undefined columns: { stacked: Column[] tabs: Column[] diff --git a/apps/explorer/src/lib/live-feed.ts b/apps/explorer/src/lib/live-feed.ts new file mode 100644 index 000000000..c85fe0d60 --- /dev/null +++ b/apps/explorer/src/lib/live-feed.ts @@ -0,0 +1,9 @@ +export function liveFeedSecondsLeft( + session: { scope: string; deadline: number } | null, + scope: string, + eligible: boolean, + now: number, +): number { + if (!eligible || !session || session.scope !== scope) return 0 + return Math.max(0, Math.ceil((session.deadline - now) / 1_000)) +} diff --git a/apps/explorer/src/lib/live-query.ts b/apps/explorer/src/lib/live-query.ts new file mode 100644 index 000000000..6ee0490eb --- /dev/null +++ b/apps/explorer/src/lib/live-query.ts @@ -0,0 +1,48 @@ +import * as z from 'zod/mini' +import { zAddress } from './zod' + +export const liveQuerySchema = z.object({ + address: zAddress({ lowercase: true }), + kind: z.enum(['token-transfers', 'account-transfers', 'transactions']), + limit: z.prefault(z.coerce.number().check(z.int(), z.gte(1), z.lte(200)), 10), + account: z.optional(zAddress({ lowercase: true })), + direction: z.optional(z.enum(['in', 'out'])), + include: z.prefault(z.enum(['all', 'sent', 'received']), 'all'), + status: z.optional(z.enum(['success', 'reverted'])), + after: z.optional( + z.coerce.number().check(z.int(), z.gte(0), z.lte(253402300799)), + ), + hideSubmitBatches: z.optional(z.enum(['true', 'false'])), +}) +export type LiveQuery = z.infer +export function buildLiveQueries(options: LiveQuery): string[] { + const { address, kind, limit } = options + if (kind === 'transactions') { + const sides = + options.include === 'sent' + ? ['from'] + : options.include === 'received' + ? ['to'] + : ['from', 'to'] + return sides.map( + (side) => + `SELECT hash, block_num, idx FROM txs WHERE "${side}" = '${address}'${options.after ? ` AND block_timestamp >= '${new Date(options.after * 1000).toISOString()}'` : ''} ORDER BY block_num DESC, idx DESC LIMIT ${Math.min(limit, 10)}`, + ) + } + const account = kind === 'account-transfers' ? address : options.account + const filters = [ + "topic0 = '0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef'", + ] + if (kind === 'token-transfers') filters.push(`address = '${address}'`) + const sides = !account + ? [undefined] + : options.direction === 'in' + ? ['topic2'] + : options.direction === 'out' + ? ['topic1'] + : ['topic1', 'topic2'] + return sides.map( + (side) => + `SELECT address, block_num, tx_hash, log_idx, block_timestamp, topic0, topic1, topic2, data FROM logs WHERE ${[...filters, ...(side && account ? [`${side} = '0x${account.slice(2).padStart(64, '0')}'`] : [])].join(' AND ')} ORDER BY block_num DESC, log_idx DESC LIMIT ${limit}`, + ) +} diff --git a/apps/explorer/src/lib/server/address-history.ts b/apps/explorer/src/lib/server/address-history.ts index bbd4ab692..1cc69d78f 100644 --- a/apps/explorer/src/lib/server/address-history.ts +++ b/apps/explorer/src/lib/server/address-history.ts @@ -32,13 +32,14 @@ import { isZonePortalAddress } from '#lib/domain/zones' export const [MAX_LIMIT, DEFAULT_LIMIT] = [10, 10] const HISTORY_TOTAL_CACHE_TTL = 60_000 const HISTORY_TOTAL_CACHE_MAX_ENTRIES = 50 -const SUBMIT_BATCH_SELECTOR = toFunctionSelector( +export const SUBMIT_BATCH_SELECTOR = toFunctionSelector( getAbiItem({ abi: zonePortalAbi, name: 'submitBatch' }), ) export type EnrichedTransaction = { hash: `0x${string}` blockNumber: string + transactionIndex?: number | undefined timestamp: number from: `0x${string}` to: `0x${string}` | null @@ -312,7 +313,7 @@ async function fetchFilteredHistoryPage( * Resolves TIP-20 metadata for every token referenced by the page's event * logs (symbol/decimals for the known-event summaries). */ -async function buildTokenMetadataLookup( +export async function buildTokenMetadataLookup( rows: readonly TransactionRow[], ): Promise<(address: Address.Address) => Metadata | undefined> { const config = getWagmiConfig() @@ -394,6 +395,7 @@ export function toEnrichedTransaction( return { hash: row.hash, blockNumber: toHexQuantity(row.blockNumber), + transactionIndex: row.transactionIndex ?? undefined, timestamp: parseTimestamp(row.timestamp) ?? 0, from: Address.checksum(row.sender), to, diff --git a/apps/explorer/src/lib/server/address-live.ts b/apps/explorer/src/lib/server/address-live.ts new file mode 100644 index 000000000..21f56d4da --- /dev/null +++ b/apps/explorer/src/lib/server/address-live.ts @@ -0,0 +1,296 @@ +import { parseResponse } from 'hono/client' +import * as Address from 'ox/Address' +import * as Hex from 'ox/Hex' +import { decodeEventLog, parseAbiItem } from 'viem' +import { + liveQuerySchema, + buildLiveQueries, + type LiveQuery, +} from '#lib/live-query' +import { readSse } from '#lib/sse' +import { parseTimestamp } from '#lib/timestamp' +import { api } from './tempo-api' +import { serverEnv, tempoApiUrl } from './env' +import { getChainBackend } from './network' +import { + buildTokenMetadataLookup, + toEnrichedTransaction, + SUBMIT_BATCH_SELECTOR, +} from './address-history' +import { getTransactionActivities } from './transaction-activities' +import { isZonePortalAddress } from '#lib/domain/zones' + +const transferAbi = [ + parseAbiItem( + 'event Transfer(address indexed from, address indexed to, uint256 value)', + ), +] + +export async function addressLiveResponse( + request: Request, + address: string, + chainId: number, +): Promise { + const parsed = liveQuerySchema.safeParse({ + ...Object.fromEntries(new URL(request.url).searchParams), + address, + }) + if (!parsed.success) + return Response.json( + { error: 'Invalid live feed parameters' }, + { status: 400 }, + ) + const options = parsed.data + const abort = new AbortController() + const stop = () => abort.abort() + request.signal.addEventListener('abort', stop, { once: true }) + if (request.signal.aborted) stop() + const deadline = setTimeout(stop, 120_000) + const backend = getChainBackend(chainId, 'tidx') + const encoder = new TextEncoder() + const seen = new Set() + const metadata = new Map< + string, + Promise<{ + address: Address.Address + symbol: string + decimals: number + currency: string + }> + >() + const stream = new ReadableStream({ + start(controller) { + const emit = (event: string, data: unknown) => { + if (!abort.signal.aborted) + controller.enqueue( + encoder.encode( + `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`, + ), + ) + } + const heartbeat = setInterval(() => { + if (!abort.signal.aborted) + controller.enqueue(encoder.encode(': heartbeat\n\n')) + }, 15_000) + const run = async (sql: string) => { + const url = new URL( + `${backend?.url ?? `${tempoApiUrl}/v1/indexer`}/query`, + ) + url.search = new URLSearchParams({ + sql, + chainId: String(chainId), + live: 'true', + }).toString() + const response = await fetch(url, { + signal: abort.signal, + headers: { + Accept: 'text/event-stream', + ...(backend?.headers ?? + (serverEnv.TEMPO_API_KEY + ? { 'tempo-api-key': serverEnv.TEMPO_API_KEY } + : {})), + }, + }) + if (!response.ok || !response.body) + throw new Error(`Live indexer returned ${response.status}`) + for await (const event of readSse(response.body)) { + let result: { ok: boolean; columns: string[]; rows: unknown[][] } + if (event.event === 'lagged') { + // Keep the stream open: reconnecting can repeat the same catch-up notice. + // A fresh PostgreSQL head snapshot repairs the visible window, then + // subsequent block deltas continue over the existing connection. + const snapshotUrl = new URL(url) + snapshotUrl.searchParams.delete('live') + snapshotUrl.searchParams.set('engine', 'postgres') + const snapshot = await fetch(snapshotUrl, { + signal: abort.signal, + headers: { + Accept: 'application/json', + 'Cache-Control': 'no-cache', + ...(backend?.headers ?? + (serverEnv.TEMPO_API_KEY + ? { 'tempo-api-key': serverEnv.TEMPO_API_KEY } + : {})), + }, + }) + if (!snapshot.ok) throw new Error('Live resynchronization failed') + result = (await snapshot.json()) as typeof result + } else { + if (event.event === 'error') + throw new Error(`Live indexer error: ${event.data}`) + if (event.event !== 'result') continue + result = JSON.parse(event.data) as typeof result + } + if (!result.ok) + throw new Error(`Live query failed: ${JSON.stringify(result)}`) + const rows = result.rows.map((values) => + Object.fromEntries( + result.columns.map((column, index) => [column, values[index]]), + ), + ) + const fresh = rows.filter((row) => { + const id = + options.kind === 'transactions' + ? String(row.hash) + : `${row.tx_hash}-${row.log_idx}` + if (seen.has(id)) return false + seen.add(id) + return true + }) + if (!fresh.length) continue + const data = await normalizeRows(fresh, options, chainId, metadata) + if (data.length) emit('rows', { kind: options.kind, rows: data }) + // Bound memory even for a busy account; only the latest visible window matters. + while (seen.size > 5_000) { + const first = seen.values().next().value + if (first) seen.delete(first) + } + } + } + void Promise.all(buildLiveQueries(options).map(run)) + .then(() => { + emit('end', {}) + }) + .catch((error) => { + if (!abort.signal.aborted) + console.error( + 'Address live stream failed:', + error instanceof Error + ? `${error.message} ${'detail' in error ? JSON.stringify(error.detail) : ''}` + : 'Unknown error', + ) + emit('feed-error', { message: 'Live connection interrupted' }) + }) + .finally(() => { + stop() + clearTimeout(deadline) + clearInterval(heartbeat) + request.signal.removeEventListener('abort', stop) + try { + controller.close() + } catch { + /* The client may already have cancelled. */ + } + }) + }, + cancel: stop, + }) + return new Response(stream, { + headers: { + 'Content-Type': 'text/event-stream', + 'Cache-Control': 'no-store', + 'X-Accel-Buffering': 'no', + }, + }) +} + +async function normalizeRows( + rows: Record[], + options: LiveQuery, + chainId: number, + metadata: Map< + string, + Promise<{ + address: Address.Address + symbol: string + decimals: number + currency: string + }> + >, +) { + if (options.kind === 'transactions') { + const fetched = await Promise.all( + rows.map(async (row) => { + const hash = String(row.hash) + Hex.assert(hash) + return parseResponse( + api.v1.transactions[':transactionHash'].$get({ + param: { transactionHash: hash }, + query: { chainId: String(chainId), include: 'receipt' }, + }), + ) + }), + ) + const filtered = fetched.filter((row) => { + if (options.status && row.meta?.receipt?.status !== options.status) + return false + if ( + options.hideSubmitBatches === 'true' && + isZonePortalAddress(options.address) + ) { + const calls = [ + { to: row.recipient, input: row.input, data: undefined }, + ...((row.meta?.rpc?.calls ?? []) as { + to?: string + input?: string + data?: string + }[]), + ] + if ( + calls.some( + (call) => + call.to?.toLowerCase() === options.address && + (call.input ?? call.data ?? '') + .toLowerCase() + .startsWith(SUBMIT_BATCH_SELECTOR), + ) + ) + return false + } + return true + }) + const lookup = await buildTokenMetadataLookup(filtered) + return Promise.all( + filtered.map(async (row) => + toEnrichedTransaction(row, { + includeKnownEvents: true, + getTokenMetadata: lookup, + activities: await getTransactionActivities(row.hash, chainId), + }), + ), + ) + } + return Promise.all( + rows.map(async (row) => { + const tokenAddress = String(row.address) + Address.assert(tokenAddress) + const hash = String(row.tx_hash) + Hex.assert(hash) + const decoded = decodeEventLog({ + abi: transferAbi, + data: String(row.data) as Hex.Hex, + topics: [ + String(row.topic0), + String(row.topic1), + String(row.topic2), + ] as [Hex.Hex, Hex.Hex, Hex.Hex], + }) + const transfer = { + id: `${hash}-${row.log_idx}`, + from: decoded.args.from, + to: decoded.args.to, + value: decoded.args.value.toString(), + transactionHash: hash, + blockNumber: String(row.block_num), + timestamp: String(parseTimestamp(row.block_timestamp) ?? ''), + } + if (options.kind === 'token-transfers') return transfer + if (!metadata.has(tokenAddress)) + metadata.set( + tokenAddress, + parseResponse( + api.v1.tokens[':token'].$get({ + param: { token: tokenAddress }, + query: { chainId: String(chainId) }, + }), + ).then((token) => ({ + address: tokenAddress, + symbol: token.symbol, + decimals: token.decimals, + currency: token.currency, + })), + ) + return { ...transfer, token: await metadata.get(tokenAddress) } + }), + ) +} diff --git a/apps/explorer/src/lib/server/token.ts b/apps/explorer/src/lib/server/token.ts index 7d25a95c7..22fec88a9 100644 --- a/apps/explorer/src/lib/server/token.ts +++ b/apps/explorer/src/lib/server/token.ts @@ -182,6 +182,7 @@ export type FetchTokenTransfersInput = z.infer< export type TokenTransfersApiResponse = { transfers: Array<{ + id: string from: Address.Address to: Address.Address value: string @@ -208,6 +209,7 @@ const FetchAccountTransfersInputSchema = z.object({ export type AccountTransfersApiResponse = { transfers: Array<{ + id: string from: Address.Address to: Address.Address value: string @@ -262,6 +264,7 @@ export const fetchAccountTransfers = createServerFn({ method: 'POST' }) return { transfers: page.data.map((transfer) => ({ + id: transfer.id, from: transfer.sender as Address.Address, to: transfer.recipient as Address.Address, value: transfer.sourceAmount.baseUnits, @@ -316,6 +319,7 @@ export const fetchTransfers = createServerFn({ method: 'POST' }) return { transfers: page.data.map((transfer) => ({ + id: transfer.id, from: transfer.sender as Address.Address, to: transfer.recipient as Address.Address, value: transfer.sourceAmount.baseUnits, diff --git a/apps/explorer/src/lib/sse.ts b/apps/explorer/src/lib/sse.ts new file mode 100644 index 000000000..48123abfb --- /dev/null +++ b/apps/explorer/src/lib/sse.ts @@ -0,0 +1,54 @@ +/** Read SSE frames without assuming network chunks align with events. */ +export async function* readSse(body: ReadableStream) { + const reader = body.getReader() + const decoder = new TextDecoder() + let buffer = '' + try { + while (true) { + const { value, done } = await reader.read() + buffer += done + ? decoder.decode() + : decoder.decode(value, { stream: true }) + while (true) { + const match = /\r?\n\r?\n/.exec(buffer) + if (!match) break + const frame = buffer.slice(0, match.index) + buffer = buffer.slice(match.index + match[0].length) + let event = 'message' + const data: string[] = [] + for (const line of frame.split(/\r?\n/)) { + if (line.startsWith('event:')) event = line.slice(6).trim() + if (line.startsWith('data:')) + data.push(line.slice(5).replace(/^ /, '')) + } + if (data.length) yield { event, data: data.join('\n') } + } + if (done) break + } + } finally { + await reader.cancel().catch(() => {}) + reader.releaseLock() + } +} + +export function mergeLiveRows< + T extends { + blockNumber: string + transactionIndex?: number | undefined + id?: string | undefined + }, +>(previous: T[], incoming: T[], key: (row: T) => string, limit: number): T[] { + const rows = new Map() + for (const row of [...incoming, ...previous]) + if (!rows.has(key(row))) rows.set(key(row), row) + return [...rows.values()] + .sort((a, b) => { + const blockA = BigInt(a.blockNumber), + blockB = BigInt(b.blockNumber) + if (blockA !== blockB) return blockA > blockB ? -1 : 1 + const position = (row: T) => + row.transactionIndex ?? Number(row.id?.split('-').at(-1) ?? 0) + return position(b) - position(a) + }) + .slice(0, limit) +} diff --git a/apps/explorer/src/lib/use-address-live.ts b/apps/explorer/src/lib/use-address-live.ts new file mode 100644 index 000000000..726aa2d10 --- /dev/null +++ b/apps/explorer/src/lib/use-address-live.ts @@ -0,0 +1,74 @@ +import * as React from 'react' +import type { EnrichedTransaction } from './server/address-history' +import type { + AccountTransfersApiResponse, + TokenTransfersApiResponse, +} from './server/token' + +export type LiveRows = + | { kind: 'transactions'; rows: EnrichedTransaction[] } + | { kind: 'token-transfers'; rows: TokenTransfersApiResponse['transfers'] } + | { + kind: 'account-transfers' + rows: AccountTransfersApiResponse['transfers'] + } + +export function useAddressLive( + url: string, + enabled: boolean, + onRows: (data: LiveRows) => void, + stop: () => void, +) { + const callbacks = React.useRef({ onRows, stop }) + React.useEffect(() => { + callbacks.current = { onRows, stop } + }, [onRows, stop]) + const [error, setError] = React.useState(null) + React.useEffect(() => { + if (!enabled) return + setError(null) + let source: EventSource | undefined + let retry: ReturnType | undefined + let retries = 0 + let disposed = false + const reconnect = () => { + source?.close() + if (disposed || retry) return + if (++retries > 3) { + setError('Live updates paused. Try again.') + callbacks.current.stop() + return + } + retry = setTimeout( + () => { + retry = undefined + connect() + }, + 1_000 * 2 ** (retries - 1), + ) + } + const connect = () => { + if (disposed) return + source = new EventSource(url) + source.addEventListener('rows', (event) => { + try { + callbacks.current.onRows( + JSON.parse((event as MessageEvent).data) as LiveRows, + ) + } catch { + reconnect() + } + }) + source.addEventListener('feed-error', reconnect) + source.addEventListener('end', reconnect) + source.onerror = reconnect + } + connect() + return () => { + disposed = true + source?.close() + clearTimeout(retry) + } + }, [url, enabled]) + return error +} diff --git a/apps/explorer/src/lib/use-live-feed.ts b/apps/explorer/src/lib/use-live-feed.ts new file mode 100644 index 000000000..a7b67bfad --- /dev/null +++ b/apps/explorer/src/lib/use-live-feed.ts @@ -0,0 +1,54 @@ +import * as React from 'react' +import { liveFeedSecondsLeft } from './live-feed' + +/** A live session belongs to one address, tab, page, and set of filters. */ +export function useLiveFeed( + scope: string, + eligible: boolean, + initialLive = false, +) { + const [session, setSession] = React.useState<{ + scope: string + deadline: number + } | null>(null) + const [now, setNow] = React.useState(0) + const initialized = React.useRef(false) + const stop = React.useCallback(() => setSession(null), []) + const start = React.useCallback(() => { + if (!eligible || document.visibilityState !== 'visible') return + const now = Date.now() + setNow(now) + setSession({ scope, deadline: now + 120_000 }) + }, [scope, eligible]) + React.useEffect(() => { + if (initialized.current) return + initialized.current = true + if (initialLive) start() + }, [initialLive, start]) + const secondsLeft = liveFeedSecondsLeft(session, scope, eligible, now) + const live = secondsLeft > 0 + React.useEffect(() => { + if (!session) return + if (!live) { + stop() + return + } + const tick = () => { + const now = Date.now() + setNow(now) + if (now >= session.deadline) stop() + } + const hide = () => { + if (document.visibilityState !== 'visible') stop() + } + const timer = window.setInterval(tick, 1_000) + document.addEventListener('visibilitychange', hide) + window.addEventListener('pagehide', stop) + return () => { + window.clearInterval(timer) + document.removeEventListener('visibilitychange', hide) + window.removeEventListener('pagehide', stop) + } + }, [session, live, stop]) + return { live, start, stop } +} diff --git a/apps/explorer/src/lib/use-new-live-rows.ts b/apps/explorer/src/lib/use-new-live-rows.ts new file mode 100644 index 000000000..872cd69f4 --- /dev/null +++ b/apps/explorer/src/lib/use-new-live-rows.ts @@ -0,0 +1,31 @@ +import * as React from 'react' + +type Key = string | number + +/** Track row identity across refreshes, without flashing the initial page or resumed feed. */ +export function useNewLiveRows( + keys: (Key | undefined)[], + scope: string | undefined, +) { + const signature = JSON.stringify(keys) + const [previous, setPrevious] = React.useState(() => ({ + signature, + scope, + keys: new Set(keys), + added: new Set(), + })) + if (previous.signature !== signature || previous.scope !== scope) { + const added = new Set() + if ( + scope !== undefined && + previous.scope === scope && + previous.keys.size > 0 + ) { + for (const key of keys) + if (key !== undefined && !previous.keys.has(key)) added.add(key) + } + setPrevious({ signature, scope, keys: new Set(keys), added }) + return added + } + return previous.added +} diff --git a/apps/explorer/src/routeTree.gen.ts b/apps/explorer/src/routeTree.gen.ts index df613e962..5fc9c1574 100644 --- a/apps/explorer/src/routeTree.gen.ts +++ b/apps/explorer/src/routeTree.gen.ts @@ -41,6 +41,7 @@ import { Route as ApiTokenLogoAddressRouteImport } from './routes/api/token/logo import { Route as ApiContractCreationAddressRouteImport } from './routes/api/contract/creation/$address' import { Route as ApiAddressZonePortalAddressRouteImport } from './routes/api/address/zone-portal/$address' import { Route as ApiAddressMetadataAddressRouteImport } from './routes/api/address/metadata/$address' +import { Route as ApiAddressLiveAddressRouteImport } from './routes/api/address/live.$address' import { Route as ApiAddressHistoryAddressRouteImport } from './routes/api/address/history/$address' import { Route as ApiAddressBalancesAddressRouteImport } from './routes/api/address/balances/$address' import { Route as LayoutBlockCountdownTargetBlockRouteImport } from './routes/_layout/block/countdown.$targetBlock' @@ -208,6 +209,11 @@ const ApiAddressMetadataAddressRoute = path: '/api/address/metadata/$address', getParentRoute: () => rootRouteImport, } as any) +const ApiAddressLiveAddressRoute = ApiAddressLiveAddressRouteImport.update({ + id: '/api/address/live/$address', + path: '/api/address/live/$address', + getParentRoute: () => rootRouteImport, +} as any) const ApiAddressHistoryAddressRoute = ApiAddressHistoryAddressRouteImport.update({ id: '/api/address/history/$address', @@ -262,6 +268,7 @@ export interface FileRoutesByFullPath { '/block/countdown/$targetBlock': typeof LayoutBlockCountdownTargetBlockRoute '/api/address/balances/$address': typeof ApiAddressBalancesAddressRoute '/api/address/history/$address': typeof ApiAddressHistoryAddressRoute + '/api/address/live/$address': typeof ApiAddressLiveAddressRoute '/api/address/metadata/$address': typeof ApiAddressMetadataAddressRoute '/api/address/zone-portal/$address': typeof ApiAddressZonePortalAddressRouteWithChildren '/api/contract/creation/$address': typeof ApiContractCreationAddressRoute @@ -299,6 +306,7 @@ export interface FileRoutesByTo { '/block/countdown/$targetBlock': typeof LayoutBlockCountdownTargetBlockRoute '/api/address/balances/$address': typeof ApiAddressBalancesAddressRoute '/api/address/history/$address': typeof ApiAddressHistoryAddressRoute + '/api/address/live/$address': typeof ApiAddressLiveAddressRoute '/api/address/metadata/$address': typeof ApiAddressMetadataAddressRoute '/api/address/zone-portal/$address': typeof ApiAddressZonePortalAddressRouteWithChildren '/api/contract/creation/$address': typeof ApiContractCreationAddressRoute @@ -338,6 +346,7 @@ export interface FileRoutesById { '/_layout/block/countdown/$targetBlock': typeof LayoutBlockCountdownTargetBlockRoute '/api/address/balances/$address': typeof ApiAddressBalancesAddressRoute '/api/address/history/$address': typeof ApiAddressHistoryAddressRoute + '/api/address/live/$address': typeof ApiAddressLiveAddressRoute '/api/address/metadata/$address': typeof ApiAddressMetadataAddressRoute '/api/address/zone-portal/$address': typeof ApiAddressZonePortalAddressRouteWithChildren '/api/contract/creation/$address': typeof ApiContractCreationAddressRoute @@ -377,6 +386,7 @@ export interface FileRouteTypes { | '/block/countdown/$targetBlock' | '/api/address/balances/$address' | '/api/address/history/$address' + | '/api/address/live/$address' | '/api/address/metadata/$address' | '/api/address/zone-portal/$address' | '/api/contract/creation/$address' @@ -414,6 +424,7 @@ export interface FileRouteTypes { | '/block/countdown/$targetBlock' | '/api/address/balances/$address' | '/api/address/history/$address' + | '/api/address/live/$address' | '/api/address/metadata/$address' | '/api/address/zone-portal/$address' | '/api/contract/creation/$address' @@ -452,6 +463,7 @@ export interface FileRouteTypes { | '/_layout/block/countdown/$targetBlock' | '/api/address/balances/$address' | '/api/address/history/$address' + | '/api/address/live/$address' | '/api/address/metadata/$address' | '/api/address/zone-portal/$address' | '/api/contract/creation/$address' @@ -474,6 +486,7 @@ export interface RootRouteChildren { ApiAbiBatchRoute: typeof ApiAbiBatchRoute ApiAddressBalancesAddressRoute: typeof ApiAddressBalancesAddressRoute ApiAddressHistoryAddressRoute: typeof ApiAddressHistoryAddressRoute + ApiAddressLiveAddressRoute: typeof ApiAddressLiveAddressRoute ApiAddressMetadataAddressRoute: typeof ApiAddressMetadataAddressRoute ApiAddressZonePortalAddressRoute: typeof ApiAddressZonePortalAddressRouteWithChildren ApiContractCreationAddressRoute: typeof ApiContractCreationAddressRoute @@ -708,6 +721,13 @@ declare module '@tanstack/react-router' { preLoaderRoute: typeof ApiAddressMetadataAddressRouteImport parentRoute: typeof rootRouteImport } + '/api/address/live/$address': { + id: '/api/address/live/$address' + path: '/api/address/live/$address' + fullPath: '/api/address/live/$address' + preLoaderRoute: typeof ApiAddressLiveAddressRouteImport + parentRoute: typeof rootRouteImport + } '/api/address/history/$address': { id: '/api/address/history/$address' path: '/api/address/history/$address' @@ -809,6 +829,7 @@ const rootRouteChildren: RootRouteChildren = { ApiAbiBatchRoute: ApiAbiBatchRoute, ApiAddressBalancesAddressRoute: ApiAddressBalancesAddressRoute, ApiAddressHistoryAddressRoute: ApiAddressHistoryAddressRoute, + ApiAddressLiveAddressRoute: ApiAddressLiveAddressRoute, ApiAddressMetadataAddressRoute: ApiAddressMetadataAddressRoute, ApiAddressZonePortalAddressRoute: ApiAddressZonePortalAddressRouteWithChildren, diff --git a/apps/explorer/src/routes/_layout/address/$address.tsx b/apps/explorer/src/routes/_layout/address/$address.tsx index c3dc9cf25..c28c71565 100644 --- a/apps/explorer/src/routes/_layout/address/$address.tsx +++ b/apps/explorer/src/routes/_layout/address/$address.tsx @@ -53,6 +53,9 @@ import { } from '#lib/domain/known-event-totals' import { TransactionFilters, TransferFilters } from '#comps/TransactionFilters' import { cx } from '#lib/css' +import { useLiveFeed } from '#lib/use-live-feed' +import { useAddressLive } from '#lib/use-address-live' +import { mergeLiveRows } from '#lib/sse' import { type AssetData, type BalancesResponse, @@ -1162,6 +1165,25 @@ function SectionsWrapper(props: { [address, after, include, status, hideSubmitBatches], ) + const feedScope = JSON.stringify([ + address, + activeTab, + page, + cursor, + order, + limit, + account, + status, + transferDirection, + dir, + period, + hideSubmitBatches, + ]) + const feedEligible = isTransactionsTabActive + ? isLatestHistoryPosition + : isTransfersTabActive && page === 1 + const feed = useLiveFeed(feedScope, feedEligible, live) + const latestHistoryQuery = useQuery({ ...historyQueryOptions({ address, @@ -1175,12 +1197,9 @@ function SectionsWrapper(props: { initialData, enabled: isMounted && (isTransactionsTabActive || initialData !== undefined), - refetchInterval: - live && isTransactionsTabActive && isLatestHistoryPosition - ? 4_000 - : false, - refetchOnWindowFocus: - live && isTransactionsTabActive && isLatestHistoryPosition, + refetchInterval: false, + refetchOnWindowFocus: false, + retry: false, }) const latestHistoryData = latestHistoryQuery.data const total = latestHistoryData?.total ?? undefined @@ -1213,6 +1232,9 @@ function SectionsWrapper(props: { account, }), enabled: isMounted && isToken && isTransfersTabActive, + refetchInterval: false, + refetchOnWindowFocus: false, + retry: false, }) // Account-scoped transfers query (non-token addresses): the D2 split moved @@ -1231,8 +1253,167 @@ function SectionsWrapper(props: { direction: transferDirection, }), enabled: isMounted && !isToken && isTransfersTabActive, + refetchInterval: false, + refetchOnWindowFocus: false, + retry: false, }) + const liveParams = new URLSearchParams({ + kind: isTransactionsTabActive + ? 'transactions' + : isToken + ? 'token-transfers' + : 'account-transfers', + limit: String(isTransactionsTabActive ? HISTORY_PAGE_SIZE : limit), + include, + }) + if (account) liveParams.set('account', account) + if (transferDirection) liveParams.set('direction', transferDirection) + if (status) liveParams.set('status', status) + if (after) liveParams.set('after', String(after)) + if (hideSubmitBatches) liveParams.set('hideSubmitBatches', 'true') + const streamError = useAddressLive( + `/api/address/live/${address}?${liveParams}`, + feed.live, + (update) => { + if (update.kind === 'transactions') { + queryClient.setQueryData( + getHistoryQueryOptions({ order: 'desc' }).queryKey, + (old) => { + const transactions = mergeLiveRows( + old?.transactions ?? [], + update.rows, + (row) => row.hash, + HISTORY_PAGE_SIZE, + ) + const last = transactions.at(-1) + const first = transactions[0] + const cursorFor = (row: typeof last) => + row?.transactionIndex !== undefined + ? btoa( + JSON.stringify([ + Number(BigInt(row.blockNumber)), + row.transactionIndex, + ]), + ) + : undefined + return { + total: null, + limit: HISTORY_PAGE_SIZE, + nextCursor: null, + reverseCursor: null, + countCapped: false, + ...old, + transactions, + error: null, + ...(cursorFor(last) + ? { nextCursor: cursorFor(last) ?? null } + : {}), + ...(cursorFor(first) + ? { reverseCursor: cursorFor(first) ?? null } + : {}), + } + }, + ) + } else if (update.kind === 'token-transfers') { + queryClient.setQueryData( + transfersQueryOptions({ address, page: 1, limit, account }).queryKey, + (old) => ({ + total: 0, + totalCapped: true, + ...old, + transfers: mergeLiveRows( + old?.transfers ?? [], + update.rows, + (row) => row.id, + limit, + ), + }), + ) + } else { + queryClient.setQueryData( + accountTransfersQueryOptions({ + account: address, + page: 1, + limit, + direction: transferDirection, + }).queryKey, + (old) => ({ + total: 0, + totalCapped: true, + ...old, + transfers: mergeLiveRows( + old?.transfers ?? [], + update.rows, + (row) => row.id, + limit, + ), + }), + ) + } + }, + feed.stop, + ) + + const feedError = isTransactionsTabActive + ? latestHistoryQuery.error || latestHistoryQuery.data?.error + : isToken + ? tokenTransfersError + : accountTransfersError + React.useEffect(() => { + if (feedError) feed.stop() + }, [feedError, feed.stop]) + + const liveControl = ( + <> + {streamError && ( + + {streamError} + + )} + + + ) + const { transfers = [], total: transfersTotal = 0, @@ -1871,18 +2052,22 @@ function SectionsWrapper(props: { totalItems: totalTrxCount ?? transactions.length, itemsLabel: 'transactions', contextual: ( - +
+ + {liveControl} +
), content: transactionsError ?? ( + {filters} + {liveControl} + + ), content: (

@@ -2094,9 +2284,15 @@ function SectionsWrapper(props: { accountTransfersData && (accountTotalCapped ? '10k+' : accountTotal), itemsLabel: 'transfers', - contextual: filters, + contextual: ( +

+ {filters} + {liveControl} +
+ ), content: ( + contextual: ( +
+ {account && ( + + )} + {liveControl} +
), content: ( ({ + key: transfer.id, cells: [ + addressLiveResponse( + request, + params.address, + getChainId(getWagmiConfig()), + ), + }, + }, +}) diff --git a/apps/explorer/test/address-history.node.test.ts b/apps/explorer/test/address-history.node.test.ts index 84365c169..0521637cf 100644 --- a/apps/explorer/test/address-history.node.test.ts +++ b/apps/explorer/test/address-history.node.test.ts @@ -173,6 +173,7 @@ describe('toEnrichedTransaction', () => { expect(result).toEqual({ hash: HASH, blockNumber: '0x14d5b48', + transactionIndex: 17, timestamp: Date.parse('2026-06-12T03:58:20.000Z') / 1000, from: '0x286ad6cfc7279C8a6D86D15dcEFcB77A65Aa7E92', to: '0x20C0000000000000000000000000000000000003', diff --git a/apps/explorer/test/address-live.node.test.ts b/apps/explorer/test/address-live.node.test.ts new file mode 100644 index 000000000..097986030 --- /dev/null +++ b/apps/explorer/test/address-live.node.test.ts @@ -0,0 +1,93 @@ +import { describe, expect, it } from 'vitest' +import { readSse, mergeLiveRows } from '../src/lib/sse' +import { buildLiveQueries, liveQuerySchema } from '../src/lib/live-query' +const address = '0x20c000000000000000000000b9537d11c60e8b50' +describe('address live streams', () => { + it('parses chunked CRLF frames, multiline data and lag notifications', async () => { + const bytes = new TextEncoder().encode( + ': heartbeat\r\n\r\nevent: rows\r\ndata: {"rows":\r\ndata: []}\r\n\r\nevent: lagged\ndata: {"skipped":2}\n\n', + ) + const stream = new ReadableStream({ + start(c) { + for (const byte of bytes) c.enqueue(new Uint8Array([byte])) + c.close() + }, + }) + const frames = [] + for await (const frame of readSse(stream)) frames.push(frame) + expect(frames).toEqual([ + { event: 'rows', data: '{"rows":\n[]}' }, + { event: 'lagged', data: '{"skipped":2}' }, + ]) + }) + it('cancels the reader on disconnect', async () => { + let cancelled = false + const stream = new ReadableStream({ + start(c) { + c.enqueue(new TextEncoder().encode('event: rows\ndata: []\n\n')) + }, + cancel() { + cancelled = true + }, + }) + for await (const _frame of readSse(stream)) break + expect(cancelled).toBe(true) + }) + it('deduplicates reconnect snapshots and preserves distinct logs', () => { + const old = [ + { id: 'a-1', blockNumber: '100' }, + { id: 'a-0', blockNumber: '100' }, + ] + const fresh = [{ id: 'b-0', blockNumber: '101' }, ...old] + expect(mergeLiveRows(old, fresh, (row) => row.id, 3)).toEqual(fresh) + expect(mergeLiveRows(fresh, fresh, (row) => row.id, 2)).toEqual( + fresh.slice(0, 2), + ) + }) + it('rejects invalid addresses and unbounded limits', () => { + expect( + liveQuerySchema.safeParse({ + address: "x' OR 1=1", + kind: 'token-transfers', + }).success, + ).toBe(false) + expect( + liveQuerySchema.safeParse({ + address, + kind: 'token-transfers', + limit: 10000, + }).success, + ).toBe(false) + const query = buildLiveQueries( + liveQuerySchema.parse({ address, kind: 'token-transfers', limit: 10 }), + )[0] + expect(query).toContain(`address = '${address}'`) + expect(query).toContain('topic0 =') + expect(query).toContain('LIMIT 10') + }) + it('uses separate indexed sides and honors direction', () => { + expect( + buildLiveQueries( + liveQuerySchema.parse({ address, kind: 'account-transfers' }), + ), + ).toHaveLength(2) + expect( + buildLiveQueries( + liveQuerySchema.parse({ + address, + kind: 'account-transfers', + direction: 'in', + }), + )[0], + ).toContain('topic2 =') + expect( + buildLiveQueries( + liveQuerySchema.parse({ + address, + kind: 'transactions', + include: 'sent', + }), + )[0], + ).toContain(`"from" = '${address}'`) + }) +}) diff --git a/apps/explorer/test/live-feed.node.test.ts b/apps/explorer/test/live-feed.node.test.ts new file mode 100644 index 000000000..0970f52db --- /dev/null +++ b/apps/explorer/test/live-feed.node.test.ts @@ -0,0 +1,21 @@ +import { describe, expect, it } from 'vitest' +import { liveFeedSecondsLeft } from '../src/lib/live-feed' + +describe('bounded live feed', () => { + const session = { scope: 'address:transfers:1', deadline: 120_000 } + it('is off until explicitly started', () => { + expect(liveFeedSecondsLeft(null, session.scope, true, 0)).toBe(0) + }) + it('expires after two minutes, including after a suspended timer', () => { + expect(liveFeedSecondsLeft(session, session.scope, true, 0)).toBe(120) + expect(liveFeedSecondsLeft(session, session.scope, true, 119_001)).toBe(1) + expect(liveFeedSecondsLeft(session, session.scope, true, 120_000)).toBe(0) + expect(liveFeedSecondsLeft(session, session.scope, true, 180_000)).toBe(0) + }) + it('stops when the address, tab, page, or filters change', () => { + expect( + liveFeedSecondsLeft(session, 'address:transactions:1', true, 1000), + ).toBe(0) + expect(liveFeedSecondsLeft(session, session.scope, false, 1000)).toBe(0) + }) +})