diff --git a/.env.example b/.env.example index f566e35..9640276 100644 --- a/.env.example +++ b/.env.example @@ -30,6 +30,9 @@ NEXT_PUBLIC_VIBENET_RPC_URL=https://rpc.vibes.base.org # One deployment serves all chains; provide a set per chain # (TIPS_MAINNET_, TIPS_SEPOLIA_, TIPS_ZERONET_). Secrets → Vercel env, not git. # TIPS_MAINNET_RPC_URL= +# Execution WebSocket for the live newHeads block stream. Optional per chain; +# without it the block list polls TIPS__RPC_URL. +# TIPS_MAINNET_WS_URL= # Audit events JSON-RPC endpoint (Postgres-backed transaction observability). # Opt-in per chain: the only source for bundle and rejected views, and the # enrichment source for block and transaction views. diff --git a/app/api/internal-explorer/block-list.test.ts b/app/api/internal-explorer/block-list.test.ts index 9fa2198..5703aae 100644 --- a/app/api/internal-explorer/block-list.test.ts +++ b/app/api/internal-explorer/block-list.test.ts @@ -2,7 +2,45 @@ import assert from 'node:assert/strict'; import { describe, test } from 'vitest'; -import { blockNumbersForPage, nextBlockCursor, parseBlockListQuery } from './block-list'; +import { + blockNumbersForPage, + nextBlockCursor, + parseBlockListQuery, + parseBlockSummary, +} from './block-list'; + +describe('parseBlockSummary', () => { + const rpcBlock = { + hash: `0x${'ab'.repeat(32)}`, + number: '0x10', + timestamp: '0x6b49d200', + transactions: [`0x${'01'.repeat(32)}`, `0x${'02'.repeat(32)}`], + gasUsed: '0x5208', + gasLimit: '0x7270e00', + baseFeePerGas: '0x1', + }; + + test('reads the Denim millisecond timestamp when present', () => { + const summary = parseBlockSummary({ ...rpcBlock, timestampMs: '0x1a3185c50c8' }); + assert.equal(summary?.timestampMs, 1_800_000_000_200); + assert.equal(summary?.timestamp, 1_800_000_000); + assert.equal(summary?.transactionCount, 2); + }); + + test('leaves timestampMs null for blocks without it', () => { + assert.equal(parseBlockSummary(rpcBlock)?.timestampMs, null); + }); + + test('leaves transactionCount null for a header without a transaction list', () => { + const { transactions: _transactions, ...header } = rpcBlock; + assert.equal(parseBlockSummary(header)?.transactionCount, null); + }); + + test('rejects malformed blocks', () => { + assert.equal(parseBlockSummary(null), null); + assert.equal(parseBlockSummary({ ...rpcBlock, number: undefined }), null); + }); +}); describe('block list pagination', () => { test('starts at the inclusive cursor', () => { diff --git a/app/api/internal-explorer/block-list.ts b/app/api/internal-explorer/block-list.ts index 7410277..5dc56d1 100644 --- a/app/api/internal-explorer/block-list.ts +++ b/app/api/internal-explorer/block-list.ts @@ -11,7 +11,10 @@ export interface BlockSummary { hash: string; number: number; timestamp: number; - transactionCount: number; + /** Unix milliseconds from Denim's `timestampMs`; null on blocks or nodes without it. */ + timestampMs: number | null; + /** Null when unknown, such as a streamed head whose count lookup failed. */ + transactionCount: number | null; gasUsed: string | null; gasLimit: string | null; baseFeePerGas: string | null; @@ -95,12 +98,12 @@ async function fetchLatestBlockNumber(rpcUrl: string): Promise { return hexToNumber(result); } -async function fetchBlockByNumber( - rpcUrl: string, - blockNumber: number, -): Promise { - const result = await rpcCall(rpcUrl, 'eth_getBlockByNumber', [toBlockTag(blockNumber), false]); - +/** + * Map a raw `eth_getBlockBy*` result (hashes only) or a newHeads header to a + * summary, or null if malformed. Headers carry no transaction list, so their + * `transactionCount` is null. + */ +export function parseBlockSummary(result: unknown): BlockSummary | null { if (!result || typeof result !== 'object') return null; const block = result as Record; const number = hexToNumber(block.number); @@ -113,13 +116,30 @@ async function fetchBlockByNumber( hash: String(block.hash), number, timestamp, - transactionCount: Array.isArray(block.transactions) ? block.transactions.length : 0, + timestampMs: hexToNumber(block.timestampMs), + transactionCount: Array.isArray(block.transactions) ? block.transactions.length : null, gasUsed: decimalQuantity(block.gasUsed), gasLimit: decimalQuantity(block.gasLimit), baseFeePerGas: decimalQuantity(block.baseFeePerGas), }; } +export async function fetchTransactionCountByHash( + rpcUrl: string, + hash: string, +): Promise { + return hexToNumber(await rpcCall(rpcUrl, 'eth_getBlockTransactionCountByHash', [hash])); +} + +async function fetchBlockByNumber( + rpcUrl: string, + blockNumber: number, +): Promise { + return parseBlockSummary( + await rpcCall(rpcUrl, 'eth_getBlockByNumber', [toBlockTag(blockNumber), false]), + ); +} + export async function fetchBlocksByNumbers( rpcUrl: string, numbers: number[], diff --git a/app/api/internal-explorer/config.ts b/app/api/internal-explorer/config.ts index 661bceb..b581674 100644 --- a/app/api/internal-explorer/config.ts +++ b/app/api/internal-explorer/config.ts @@ -19,6 +19,12 @@ export function getRpcUrl(chain: ExplorerChain): string { return envValue([`TIPS_${ENV_PREFIX[chain]}_RPC_URL`]) ?? 'http://localhost:8545'; } +// Execution WebSocket endpoint used for the live newHeads stream. Opt-in per +// chain via TIPS__WS_URL; when unset the block list polls instead. +export function getWsUrl(chain: ExplorerChain): string | undefined { + return envValue([`TIPS_${ENV_PREFIX[chain]}_WS_URL`]); +} + // Audit events JSON-RPC endpoint (Postgres-backed) for a chain. Unlike the // execution RPC there is no default: audit is opt-in per chain via // TIPS__AUDIT_RPC_URL. When unset, the audit source is treated as diff --git a/app/api/internal-explorer/head-stream.test.ts b/app/api/internal-explorer/head-stream.test.ts new file mode 100644 index 0000000..95c19b3 --- /dev/null +++ b/app/api/internal-explorer/head-stream.test.ts @@ -0,0 +1,215 @@ +import assert from 'node:assert/strict'; + +import { describe, test, vi } from 'vitest'; + +import type { BlockSummary } from './block-list'; +import { HeadHub, type HeadSocket } from './head-stream'; + +const HASH = `0x${'ab'.repeat(32)}`; + +class FakeSocket implements HeadSocket { + readyState = 0; + readonly sent: unknown[] = []; + closed = false; + private readonly listeners = new Map void>>(); + + send(data: string) { + this.sent.push(JSON.parse(data)); + } + + close() { + this.closed = true; + } + + addEventListener(type: string, listener: (event?: never) => void) { + const list = this.listeners.get(type) ?? []; + list.push(listener as (event?: unknown) => void); + this.listeners.set(type, list); + } + + emit(type: 'open' | 'close' | 'error'): void; + emit(type: 'message', payload: unknown): void; + emit(type: string, payload?: unknown) { + const event = type === 'message' ? { data: JSON.stringify(payload) } : undefined; + for (const listener of this.listeners.get(type) ?? []) listener(event); + } +} + +function setup( + fetchTransactionCount: (hash: string) => Promise = async () => 7, +) { + const sockets: FakeSocket[] = []; + const onEnded = vi.fn(); + const hub = new HeadHub({ + wsUrl: 'wss://node.example:8546', + fetchTransactionCount, + onEnded, + connect: () => { + const socket = new FakeSocket(); + sockets.push(socket); + return socket; + }, + }); + return { hub, sockets, onEnded }; +} + +const headNotification = (hash: string, extra: Record = {}) => ({ + jsonrpc: '2.0', + method: 'eth_subscription', + params: { + subscription: '0x1', + result: { number: '0x10', hash, timestamp: '0x6b49d200', gasUsed: '0x5208', ...extra }, + }, +}); + +describe('HeadHub', () => { + test('shares one upstream subscription across subscribers', () => { + const { hub, sockets } = setup(); + hub.subscribe({ onBlock: () => {}, onEnd: () => {} }); + hub.subscribe({ onBlock: () => {}, onEnd: () => {} }); + sockets[0].emit('open'); + + assert.equal(sockets.length, 1); + assert.deepEqual(sockets[0].sent, [ + { jsonrpc: '2.0', id: 1, method: 'eth_subscribe', params: ['newHeads'] }, + ]); + }); + + test('delivers each head to every subscriber with its header and transaction count', async () => { + const fetchCount = vi.fn(async () => 7); + const { hub, sockets } = setup(fetchCount); + const received: BlockSummary[][] = [[], []]; + hub.subscribe({ onBlock: (block) => received[0].push(block), onEnd: () => {} }); + hub.subscribe({ onBlock: (block) => received[1].push(block), onEnd: () => {} }); + sockets[0].emit('open'); + sockets[0].emit('message', headNotification(HASH, { timestampMs: '0x1a3185c50c8' })); + + await vi.waitFor(() => assert.equal(received[1].length, 1)); + assert.deepEqual(fetchCount.mock.calls, [[HASH]]); + assert.deepEqual(received[0][0], { + hash: HASH, + number: 16, + timestamp: 1_800_000_000, + timestampMs: 1_800_000_000_200, + transactionCount: 7, + gasUsed: '21000', + gasLimit: null, + baseFeePerGas: null, + }); + }); + + test('leaves timestampMs null when the header lacks it', async () => { + const { hub, sockets } = setup(); + const received: BlockSummary[] = []; + hub.subscribe({ onBlock: (block) => received.push(block), onEnd: () => {} }); + sockets[0].emit('open'); + sockets[0].emit('message', headNotification(HASH)); + + await vi.waitFor(() => assert.equal(received.length, 1)); + assert.equal(received[0].timestampMs, null); + assert.equal(received[0].timestamp, 1_800_000_000); + }); + + test('delivers heads in arrival order even when lookups finish out of order', async () => { + const resolvers = new Map void>(); + const { hub, sockets } = setup( + (hash) => new Promise((resolve) => resolvers.set(hash, resolve)), + ); + const received: string[] = []; + hub.subscribe({ onBlock: (block) => received.push(block.hash), onEnd: () => {} }); + sockets[0].emit('open'); + const first = `0x${'01'.repeat(32)}`; + const second = `0x${'02'.repeat(32)}`; + sockets[0].emit('message', headNotification(first)); + sockets[0].emit('message', headNotification(second)); + + resolvers.get(second)?.(1); + await new Promise((resolve) => setTimeout(resolve, 0)); + assert.deepEqual(received, []); + resolvers.get(first)?.(1); + await vi.waitFor(() => assert.deepEqual(received, [first, second])); + }); + + test('delivers a head without a count when the lookup fails or stalls', async () => { + vi.useFakeTimers(); + try { + const stalled = `0x${'03'.repeat(32)}`; + const { hub, sockets } = setup((hash) => + hash === stalled ? new Promise(() => {}) : Promise.reject(new Error('rpc down')), + ); + const received: BlockSummary[] = []; + hub.subscribe({ onBlock: (block) => received.push(block), onEnd: () => {} }); + sockets[0].emit('open'); + sockets[0].emit('message', headNotification(HASH)); + sockets[0].emit('message', headNotification(stalled)); + + await vi.advanceTimersByTimeAsync(1_000); + assert.deepEqual( + received.map((block) => block.transactionCount), + [null, null], + ); + } finally { + vi.useRealTimers(); + } + }); + + test('ignores notifications without a valid block hash', async () => { + const fetchCount = vi.fn(async () => 7); + const { hub, sockets } = setup(fetchCount); + const onBlock = vi.fn(); + hub.subscribe({ onBlock, onEnd: () => {} }); + sockets[0].emit('open'); + sockets[0].emit('message', headNotification('not-a-hash')); + sockets[0].emit('message', { jsonrpc: '2.0', id: 1, result: '0x1' }); + + await new Promise((resolve) => setTimeout(resolve, 0)); + assert.equal(fetchCount.mock.calls.length, 0); + assert.equal(onBlock.mock.calls.length, 0); + }); + + test('closes the upstream when the last subscriber leaves', () => { + const { hub, sockets, onEnded } = setup(); + const first = hub.subscribe({ onBlock: () => {}, onEnd: () => {} }); + const second = hub.subscribe({ onBlock: () => {}, onEnd: () => {} }); + + first(); + assert.equal(sockets[0].closed, false); + second(); + assert.equal(sockets[0].closed, true); + assert.equal(onEnded.mock.calls.length, 1); + }); + + test('ends every subscriber when the upstream drops', () => { + const { hub, sockets, onEnded } = setup(); + const ended = [vi.fn(), vi.fn()]; + hub.subscribe({ onBlock: () => {}, onEnd: ended[0] }); + hub.subscribe({ onBlock: () => {}, onEnd: ended[1] }); + sockets[0].emit('close'); + + assert.equal(ended[0].mock.calls.length, 1); + assert.equal(ended[1].mock.calls.length, 1); + assert.equal(onEnded.mock.calls.length, 1); + }); + + test('ends subscribers when the subscription is rejected', () => { + const { hub, sockets } = setup(); + const onEnd = vi.fn(); + hub.subscribe({ onBlock: () => {}, onEnd }); + sockets[0].emit('open'); + sockets[0].emit('message', { jsonrpc: '2.0', id: 1, error: { message: 'not supported' } }); + + assert.equal(onEnd.mock.calls.length, 1); + assert.equal(sockets[0].closed, true); + }); + + test('a subscriber added after the hub ended is ended immediately', () => { + const { hub, sockets } = setup(); + hub.subscribe({ onBlock: () => {}, onEnd: () => {} }); + sockets[0].emit('error'); + + const onEnd = vi.fn(); + hub.subscribe({ onBlock: () => {}, onEnd }); + assert.equal(onEnd.mock.calls.length, 1); + assert.equal(sockets.length, 1); + }); +}); diff --git a/app/api/internal-explorer/head-stream.ts b/app/api/internal-explorer/head-stream.ts new file mode 100644 index 0000000..0e9d6a7 --- /dev/null +++ b/app/api/internal-explorer/head-stream.ts @@ -0,0 +1,170 @@ +// Live block heads for the Internal Explorer. One upstream `eth_subscribe` +// newHeads WebSocket per endpoint is shared by every connected browser. Each +// head is summarized from its header (`timestampMs` when the upstream sends +// it) plus one transaction-count lookup, since newHeads carries no transaction +// list. Blocks are delivered complete and in arrival order. Server-only. +import { fetchTransactionCountByHash, parseBlockSummary, type BlockSummary } from './block-list'; + +const CONNECT_TIMEOUT_MS = 10_000; +// Delivery is ordered, so a stuck lookup would hold back every later head. +const COUNT_TIMEOUT_MS = 1_000; +const SUBSCRIBE_REQUEST_ID = 1; +const BLOCK_HASH = /^0x[0-9a-f]{64}$/i; + +/** The subset of the WHATWG WebSocket the hub uses; injectable for tests. */ +export interface HeadSocket { + readonly readyState: number; + send(data: string): void; + close(): void; + addEventListener(type: 'open' | 'close' | 'error', listener: () => void): void; + addEventListener(type: 'message', listener: (event: { data: unknown }) => void): void; +} + +export interface HeadSubscriber { + onBlock: (block: BlockSummary) => void; + /** Called once when the upstream stream ends; the subscriber is already removed. */ + onEnd: () => void; +} + +export interface HeadHubOptions { + wsUrl: string; + fetchTransactionCount: (hash: string) => Promise; + connect?: (url: string) => HeadSocket; + /** Called once the hub has ended, so a registry can drop it. */ + onEnded?: () => void; +} + +export class HeadHub { + private readonly subscribers = new Set(); + private socket: HeadSocket | null = null; + private connectTimer: ReturnType | undefined; + private ended = false; + private delivery: Promise = Promise.resolve(); + + constructor(private readonly options: HeadHubOptions) {} + + /** Add a subscriber, opening the upstream on first use. Returns an unsubscribe function. */ + subscribe(subscriber: HeadSubscriber): () => void { + if (this.ended) { + subscriber.onEnd(); + return () => {}; + } + this.subscribers.add(subscriber); + if (!this.socket) this.open(); + return () => { + this.subscribers.delete(subscriber); + if (this.subscribers.size === 0) this.end(); + }; + } + + private open() { + const connect = this.options.connect ?? ((url: string) => new WebSocket(url)); + let socket: HeadSocket; + try { + socket = connect(this.options.wsUrl); + } catch (error) { + console.error('Internal Explorer head stream failed to connect:', error); + this.end(); + return; + } + this.socket = socket; + this.connectTimer = setTimeout(() => this.end(), CONNECT_TIMEOUT_MS); + + socket.addEventListener('open', () => { + clearTimeout(this.connectTimer); + socket.send( + JSON.stringify({ + jsonrpc: '2.0', + id: SUBSCRIBE_REQUEST_ID, + method: 'eth_subscribe', + params: ['newHeads'], + }), + ); + }); + socket.addEventListener('message', (event) => this.handleMessage(event.data)); + socket.addEventListener('error', () => this.end()); + socket.addEventListener('close', () => this.end()); + } + + private handleMessage(data: unknown) { + let message: { + id?: unknown; + error?: unknown; + method?: unknown; + params?: { result?: Record }; + }; + try { + message = JSON.parse(String(data)); + } catch { + return; + } + + if (message.id === SUBSCRIBE_REQUEST_ID && message.error) { + console.error('Internal Explorer head stream subscription rejected:', message.error); + this.end(); + return; + } + if (message.method !== 'eth_subscription') return; + + const header = message.params?.result; + const hash = header?.hash; + if (typeof hash !== 'string' || !BLOCK_HASH.test(hash)) return; + + const block = parseBlockSummary(header); + if (!block) return; + + const count = this.lookUpTransactionCount(hash); + this.delivery = this.delivery.then(async () => { + const transactionCount = await count; + if (this.ended) return; + for (const subscriber of this.subscribers) subscriber.onBlock({ ...block, transactionCount }); + }); + } + + private lookUpTransactionCount(hash: string): Promise { + let timer: ReturnType | undefined; + const timeout = new Promise((resolve) => { + timer = setTimeout(() => resolve(null), COUNT_TIMEOUT_MS); + }); + const lookup = this.options.fetchTransactionCount(hash).catch(() => null); + return Promise.race([lookup, timeout]).finally(() => clearTimeout(timer)); + } + + private end() { + if (this.ended) return; + this.ended = true; + clearTimeout(this.connectTimer); + try { + this.socket?.close(); + } catch { + // Already closed. + } + const subscribers = [...this.subscribers]; + this.subscribers.clear(); + this.options.onEnded?.(); + for (const subscriber of subscribers) subscriber.onEnd(); + } +} + +const hubs = new Map(); + +/** Subscribe to live heads for a chain, sharing one upstream socket per WebSocket URL. */ +export function subscribeToHeads( + wsUrl: string, + rpcUrl: string, + subscriber: HeadSubscriber, +): () => void { + let hub = hubs.get(wsUrl); + if (!hub) { + const created = new HeadHub({ + wsUrl, + fetchTransactionCount: (hash) => fetchTransactionCountByHash(rpcUrl, hash), + onEnded: () => { + if (hubs.get(wsUrl) === created) hubs.delete(wsUrl); + }, + }); + hubs.set(wsUrl, created); + hub = created; + } + return hub.subscribe(subscriber); +} diff --git a/app/api/internal-explorer/heads/route.ts b/app/api/internal-explorer/heads/route.ts new file mode 100644 index 0000000..1ba367e --- /dev/null +++ b/app/api/internal-explorer/heads/route.ts @@ -0,0 +1,70 @@ +import { resolveExplorerChainFromRequest } from '../chain'; +import { getRpcUrl, getWsUrl } from '../config'; +import { explorerDisabledResponse } from '../guard'; +import { subscribeToHeads } from '../head-stream'; + +export const runtime = 'nodejs'; +export const dynamic = 'force-dynamic'; + +// Comment frames keep idle proxies from closing the stream between heads. +const KEEPALIVE_MS = 15_000; +// EventSource waits this long before reconnecting after the stream ends. +const RECONNECT_MS = 3_000; + +// Server-Sent Events stream of new canonical blocks (BlockSummary JSON per +// `data:` frame), fed by the chain's newHeads WebSocket. 404 when the chain +// has no TIPS__WS_URL, so the client falls back to polling /blocks. +export async function GET(request: Request) { + const disabled = explorerDisabledResponse(); + if (disabled) return disabled; + const chain = resolveExplorerChainFromRequest(request); + const wsUrl = getWsUrl(chain); + if (!wsUrl) { + return Response.json({ error: 'Head stream not configured' }, { status: 404 }); + } + const encoder = new TextEncoder(); + let finish = () => {}; + + const stream = new ReadableStream({ + start(controller) { + let done = false; + let unsubscribe = () => {}; + const send = (frame: string) => { + if (!done) controller.enqueue(encoder.encode(frame)); + }; + + const keepalive = setInterval(() => send(': keepalive\n\n'), KEEPALIVE_MS); + finish = () => { + if (done) return; + done = true; + clearInterval(keepalive); + request.signal.removeEventListener('abort', finish); + unsubscribe(); + try { + controller.close(); + } catch { + // Already closed by the client. + } + }; + + send(`retry: ${RECONNECT_MS}\n\n`); + request.signal.addEventListener('abort', finish); + unsubscribe = subscribeToHeads(wsUrl, getRpcUrl(chain), { + onBlock: (block) => send(`data: ${JSON.stringify(block)}\n\n`), + onEnd: () => finish(), + }); + }, + cancel() { + finish(); + }, + }); + + return new Response(stream, { + headers: { + 'Content-Type': 'text/event-stream', + 'Cache-Control': 'no-cache, no-transform', + Connection: 'keep-alive', + 'X-Accel-Buffering': 'no', + }, + }); +} diff --git a/app/internal-explorer/library/block-window.test.ts b/app/internal-explorer/library/block-window.test.ts new file mode 100644 index 0000000..da16230 --- /dev/null +++ b/app/internal-explorer/library/block-window.test.ts @@ -0,0 +1,55 @@ +import assert from 'node:assert/strict'; + +import { describe, test } from 'vitest'; + +import { mergeBlockWindow } from './block-window'; +import type { BlockSummary } from './types'; + +function block(number: number, hash = `0x${number.toString(16)}`): BlockSummary { + return { + hash, + number, + timestamp: 1_800_000_000, + timestampMs: null, + transactionCount: 2, + gasUsed: null, + gasLimit: null, + baseFeePerGas: null, + }; +} + +const numbers = (blocks: BlockSummary[]) => blocks.map((b) => b.number); + +describe('mergeBlockWindow', () => { + test('keeps the newest blocks first and trims to the limit', () => { + const current = [block(12), block(11), block(10)]; + assert.deepEqual(numbers(mergeBlockWindow(current, [block(13)], 3)), [13, 12, 11]); + }); + + test('places out-of-order heads by height', () => { + const merged = mergeBlockWindow([block(13), block(11)], [block(12)], 10); + assert.deepEqual(numbers(merged), [13, 12, 11]); + }); + + test('replaces a block at the same height with the incoming one', () => { + const merged = mergeBlockWindow([block(12, '0xold'), block(11)], [block(12, '0xnew')], 10); + assert.deepEqual( + merged.map((b) => b.hash), + ['0xnew', '0xb'], + ); + }); + + test('keeps streamed heads newer than a late polled snapshot', () => { + const streamed = [block(20), block(19)]; + const snapshot = [block(18), block(17), block(16)]; + assert.deepEqual(numbers(mergeBlockWindow(streamed, snapshot, 4)), [20, 19, 18, 17]); + }); + + test('a repeated head keeps details already filled for that block', () => { + const filled = { ...block(12), timestampMs: 1_800_000_000_400 }; + const header = { ...block(12), transactionCount: null }; + const [merged] = mergeBlockWindow([filled], [header], 10); + assert.equal(merged.transactionCount, 2); + assert.equal(merged.timestampMs, 1_800_000_000_400); + }); +}); diff --git a/app/internal-explorer/library/block-window.ts b/app/internal-explorer/library/block-window.ts new file mode 100644 index 0000000..4ebe8ca --- /dev/null +++ b/app/internal-explorer/library/block-window.ts @@ -0,0 +1,31 @@ +// Newest-first window of recent blocks shared by the live head stream and the +// polling fallback. Client-safe. +import type { BlockSummary } from './types'; + +/** + * Merge `incoming` into `current`, keeping one block per height (the incoming + * one wins, so a reorged head replaces the old block) and the newest `limit`. + * A repeat of the same block keeps details the incoming copy lacks. + */ +export function mergeBlockWindow( + current: readonly BlockSummary[], + incoming: readonly BlockSummary[], + limit: number, +): BlockSummary[] { + const byNumber = new Map(); + for (const block of current) byNumber.set(block.number, block); + for (const block of incoming) { + const existing = byNumber.get(block.number); + byNumber.set( + block.number, + existing?.hash === block.hash + ? { + ...block, + timestampMs: block.timestampMs ?? existing.timestampMs, + transactionCount: block.transactionCount ?? existing.transactionCount, + } + : block, + ); + } + return [...byNumber.values()].sort((lhs, rhs) => rhs.number - lhs.number).slice(0, limit); +} \ No newline at end of file diff --git a/app/internal-explorer/library/client.ts b/app/internal-explorer/library/client.ts index 9533372..be5f335 100644 --- a/app/internal-explorer/library/client.ts +++ b/app/internal-explorer/library/client.ts @@ -64,6 +64,8 @@ function withQuery(path: string, params: Record get('/api/internal-explorer/blocks', chain, signal), + /** Server-Sent Events URL for live BlockSummary heads; 404s when the chain has no WS URL. */ + headStreamUrl: (chain: ExplorerChain) => `/api/internal-explorer/heads?chain=${chain}`, blocksPage: ( chain: ExplorerChain, options?: { cursor?: number; limit?: number }, diff --git a/app/internal-explorer/library/format.ts b/app/internal-explorer/library/format.ts index 9f41b55..70a63dc 100644 --- a/app/internal-explorer/library/format.ts +++ b/app/internal-explorer/library/format.ts @@ -41,6 +41,13 @@ export function timeAgoFromSeconds(ts: number): string { return `${Math.floor(seconds / 86400)}d ago`; } +/** Like timeAgoFromSeconds, with tenths of a second under 10s for 200ms blocks. */ +export function timeAgoFromMilliseconds(ts: number, now = Date.now()): string { + const milliseconds = Math.max(0, Math.floor(now - ts)); + if (milliseconds < 10_000) return `${(Math.floor(milliseconds / 100) / 10).toFixed(1)}s ago`; + return timeAgoFromSeconds(ts / 1000); +} + /** `0x1234…abcd` abbreviation for a hash/address. */ export function shortHash(value: string, lead = 10, tail = 8): string { if (value.length <= lead + tail + 1) return value; diff --git a/app/internal-explorer/page.tsx b/app/internal-explorer/page.tsx index 9019239..c54cd45 100644 --- a/app/internal-explorer/page.tsx +++ b/app/internal-explorer/page.tsx @@ -24,7 +24,14 @@ import { formatSignedInteger, formatSignedPct, } from './library/explorer-format'; -import { formatGasPrice, formatHexValue, shortHash, timeAgoFromSeconds } from './library/format'; +import { mergeBlockWindow } from './library/block-window'; +import { + formatGasPrice, + formatHexValue, + shortHash, + timeAgoFromMilliseconds, + timeAgoFromSeconds, +} from './library/format'; import { explorerHref } from './library/links'; import { formatRejectionReason, @@ -162,7 +169,11 @@ function BlockRow({ #{block.number.toLocaleString()} - {timeAgoFromSeconds(block.timestamp)} + + {block.timestampMs !== null + ? timeAgoFromMilliseconds(block.timestampMs) + : timeAgoFromSeconds(block.timestamp)} +
{block.hash} @@ -170,7 +181,7 @@ function BlockRow({
- {block.transactionCount} + {block.transactionCount ?? '—'} txns
@@ -219,13 +230,16 @@ function BlocksTab({ const { showShadowDelta, setShowShadowDelta } = useShadowDelta(); const [shadowCandidates, setShadowCandidates] = useState>({}); - // Canonical-block view: poll the chain tip directly. The shadow-delta view - // has its own source below, so this fetch stands down when it is on. + // Canonical-block view: load the chain tip once, then follow the live head + // stream. Poll instead while the stream is unavailable (not configured for + // the chain, or reconnecting). The shadow-delta view has its own source + // below, so this stands down when it is on. useEffect(() => { if (showShadowDelta) return undefined; let cancelled = false; const controller = new AbortController(); + let pollTimer: number | undefined; setLoading(true); setBlocks([]); setFetchFailed(false); @@ -235,7 +249,7 @@ function BlocksTab({ try { const data = await explorerApi.blocks(chain, controller.signal); if (!cancelled) { - setBlocks(data.blocks); + setBlocks((current) => mergeBlockWindow(current, data.blocks, HOME_BLOCK_LIMIT)); setFetchFailed(false); } } catch { @@ -245,12 +259,47 @@ function BlocksTab({ } } + function startPolling() { + if (pollTimer === undefined) pollTimer = window.setInterval(() => void load(), 2000); + } + + function stopPolling() { + window.clearInterval(pollTimer); + pollTimer = undefined; + } + void load(); - const timer = window.setInterval(() => void load(), 2000); + + const source = + typeof EventSource === 'undefined' ? null : new EventSource(explorerApi.headStreamUrl(chain)); + if (source) { + source.onmessage = (event) => { + let head: BlockSummary; + try { + head = JSON.parse(event.data) as BlockSummary; + } catch { + return; + } + stopPolling(); + setBlocks((current) => mergeBlockWindow(current, [head], HOME_BLOCK_LIMIT)); + setFetchFailed(false); + setLoading(false); + }; + // EventSource retries on its own after a dropped stream; it only gives + // up (CLOSED) on a non-stream response such as the not-configured 404. + source.onerror = () => { + startPolling(); + if (source.readyState === EventSource.CLOSED) source.close(); + }; + } else { + startPolling(); + } + return () => { cancelled = true; controller.abort(); - window.clearInterval(timer); + source?.close(); + stopPolling(); }; }, [chain, showShadowDelta]);