feat(internal-explorer): stream new blocks over newHeads

.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_<CHAIN>_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.

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', () => {

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<number | null> {
   return hexToNumber(result);
 }
 
-async function fetchBlockByNumber(
-  rpcUrl: string,
-  blockNumber: number,
-): Promise<BlockSummary | null> {
-  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<string, unknown>;
   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<number | null> {
+  return hexToNumber(await rpcCall(rpcUrl, 'eth_getBlockTransactionCountByHash', [hash]));
+}
+
+async function fetchBlockByNumber(
+  rpcUrl: string,
+  blockNumber: number,
+): Promise<BlockSummary | null> {
+  return parseBlockSummary(
+    await rpcCall(rpcUrl, 'eth_getBlockByNumber', [toBlockTag(blockNumber), false]),
+  );
+}
+
 export async function fetchBlocksByNumbers(
   rpcUrl: string,
   numbers: number[],

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_<CHAIN>_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_<CHAIN>_AUDIT_RPC_URL. When unset, the audit source is treated as

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<string, Array<(event?: unknown) => 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<number | null> = 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<string, unknown> = {}) => ({
+  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<string, (count: number) => 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);
+  });
+});