mirror of
https://github.com/luxfi/explore.git
synced 2026-09-24 00:28:18 +00:00
fix(realtime): swap Phoenix WebSocket transport for backend SSE
The FE connected realtime via Phoenix Channels (wss .../socket/v2), but the
Go backend serves no Phoenix endpoint — its realtime is SSE at
/v1/base/realtime (envelope {event,chain,data}). Result: repeating ws 404 ->
'Live updates temporarily delayed'.
lib/socket/sse.ts reimplements the slice of the Phoenix Socket/Channel API the
27 realtime consumers use, backed by one EventSource, behind the existing
SocketProvider/useSocketChannel/useSocketMessage seam — no consumer changes.
Envelope routes via an optional topic field or a hub-channel translation table
(blocks -> blocks:new_block/new_block today). CT harness becomes a tiny SSE
server with the same signatures. 12 vitest cases cover it.
Frontend-only; chain/indexer untouched. Data freshness remains gated on the
separate C-Chain-rollback/indexer-reorg owner items.
This commit is contained in:
@@ -166,6 +166,38 @@ All symlinks reference this single source of truth.
|
||||
|
||||
## Recent Changes
|
||||
|
||||
### Realtime transport: Phoenix WebSocket → SSE (v1.0.13)
|
||||
- **Bug:** the FE connected realtime via Phoenix Channels
|
||||
(`wss://api-explore.lux.network/socket/v2/websocket`), but the Go backend
|
||||
(`luxfi/explorer`) serves **no** Phoenix endpoint — its realtime is **SSE**
|
||||
at `/v1/base/realtime` (multiplexed, envelope `{event, chain, data}`, initial
|
||||
`event: CONNECT`, `: ping` keep-alive). Result: repeating `ws 404` →
|
||||
"Live updates temporarily delayed". This is a frontend-only transport
|
||||
mismatch; the chain/indexer are untouched.
|
||||
- **Fix:** `lib/socket/sse.ts` reimplements the slice of the Phoenix
|
||||
`Socket`/`Channel` API the 27 realtime consumers use, backed by ONE
|
||||
`EventSource`. The seam is unchanged — `SocketProvider` / `useSocketChannel`
|
||||
/ `useSocketMessage` and every consumer keep working; only the transport
|
||||
underneath swapped. No more `phoenix` import in app code.
|
||||
- **Routing:** envelope `{event, chain, data, topic?}`. With `topic` →
|
||||
delivered straight to that channel/event (scoped `addresses:*`/`tokens:*`,
|
||||
and the component-test harness). Without → the `TRANSLATIONS` table maps the
|
||||
backend hub channel to a Blockscout `(topic, event)`; today the indexer
|
||||
broadcasts only `blocks` → `blocks:new_block` / `new_block`. `join()` resolves
|
||||
`ok` on stream open; EventSource auto-reconnects. URL normaliser collapses any
|
||||
`wss://…/socket*` (or test `ws://host:port`) to `<https>/v1/base/realtime`.
|
||||
- **Tests:** `lib/socket/sse.spec.ts` (12 vitest cases, all green) covers URL
|
||||
normalisation, blocks→new_block translation, direct-topic routing, join `ok`,
|
||||
on/off, onError, disconnect. CT harness `playwright/fixtures/socketServer.ts`
|
||||
is now a tiny SSE server (same `createSocket`/`joinChannel`/`sendMessage`
|
||||
signatures) so the 12 consumer specs are unchanged.
|
||||
- **Still owner-gated (separate, NOT this fix):** data freshness on
|
||||
explore.lux.network is blocked on the C-Chain rollback + indexer reorg
|
||||
handling — no fresh blocks flow until that lands. When it does, align the
|
||||
EVMBlock→Blockscout `Block` field shapes in the `blocks` translation (or have
|
||||
the backend emit Blockscout-shaped blocks); that is the only remaining piece
|
||||
and it can only be verified end-to-end once data flows again.
|
||||
|
||||
### Investor-grade data: real validators + honest-empty stats/DEX (v1.0.11)
|
||||
Removed every fabricated placeholder; data is now real-from-chain or honestly
|
||||
empty, never faked. Root causes + fixes:
|
||||
|
||||
@@ -38,6 +38,7 @@
|
||||
"zooai",
|
||||
"parsdao",
|
||||
"Kaniko",
|
||||
"unstub",
|
||||
"abfnrtv",
|
||||
"abitype",
|
||||
"AIVM",
|
||||
|
||||
@@ -1,8 +1,8 @@
|
||||
// https://hexdocs.pm/phoenix/js/
|
||||
import type { Channel, SocketConnectOption } from 'phoenix';
|
||||
import { Socket } from 'phoenix';
|
||||
import React, { useEffect, useState } from 'react';
|
||||
|
||||
import type { Channel, SocketOptions } from './sse';
|
||||
import { Socket } from './sse';
|
||||
|
||||
type ChannelRegistry = Record<string, { channel: Channel; subscribers: number }>;
|
||||
|
||||
const socketContexts = new Map<string, React.Context<{
|
||||
@@ -23,7 +23,7 @@ function getSocketContext(name: string) {
|
||||
interface SocketProviderProps {
|
||||
children: React.ReactNode;
|
||||
url?: string;
|
||||
options?: Partial<SocketConnectOption>;
|
||||
options?: SocketOptions;
|
||||
name?: string;
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,204 @@
|
||||
import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest';
|
||||
|
||||
import { Socket } from './sse';
|
||||
|
||||
// Controllable EventSource stand-in. Drives the adapter the way the Lux
|
||||
// explorer backend would: open the stream, push `{event,chain,data}`
|
||||
// envelopes, raise errors.
|
||||
class FakeEventSource {
|
||||
static last: FakeEventSource | undefined;
|
||||
|
||||
readonly CONNECTING = 0;
|
||||
readonly OPEN = 1;
|
||||
readonly CLOSED = 2;
|
||||
|
||||
url: string;
|
||||
withCredentials: boolean;
|
||||
readyState = 0;
|
||||
onopen: (() => void) | null = null;
|
||||
onmessage: ((event: { data: string }) => void) | null = null;
|
||||
onerror: (() => void) | null = null;
|
||||
private listeners = new Map<string, Set<() => void>>();
|
||||
|
||||
constructor(url: string, init?: { withCredentials?: boolean }) {
|
||||
this.url = url;
|
||||
this.withCredentials = init?.withCredentials ?? false;
|
||||
FakeEventSource.last = this;
|
||||
}
|
||||
|
||||
addEventListener(type: string, cb: () => void) {
|
||||
const set = this.listeners.get(type) ?? new Set();
|
||||
set.add(cb);
|
||||
this.listeners.set(type, set);
|
||||
}
|
||||
|
||||
removeEventListener(type: string, cb: () => void) {
|
||||
this.listeners.get(type)?.delete(cb);
|
||||
}
|
||||
|
||||
close() {
|
||||
this.readyState = this.CLOSED;
|
||||
}
|
||||
|
||||
// Test drivers ----------------------------------------------------------
|
||||
emitOpen() {
|
||||
this.readyState = this.OPEN;
|
||||
this.onopen?.();
|
||||
this.listeners.get('CONNECT')?.forEach((cb) => cb());
|
||||
}
|
||||
|
||||
emitMessage(envelope: object) {
|
||||
this.onmessage?.({ data: JSON.stringify(envelope) });
|
||||
}
|
||||
|
||||
emitError(closed = false) {
|
||||
if (closed) {
|
||||
this.readyState = this.CLOSED;
|
||||
}
|
||||
this.onerror?.();
|
||||
}
|
||||
}
|
||||
|
||||
function lastEs(): FakeEventSource {
|
||||
if (!FakeEventSource.last) {
|
||||
throw new Error('EventSource was not constructed');
|
||||
}
|
||||
return FakeEventSource.last;
|
||||
}
|
||||
|
||||
beforeEach(() => {
|
||||
FakeEventSource.last = undefined;
|
||||
vi.stubGlobal('EventSource', FakeEventSource);
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
vi.unstubAllGlobals();
|
||||
});
|
||||
|
||||
describe('SSE Socket — URL normalisation', () => {
|
||||
it('rewrites a Phoenix socket URL to the backend SSE endpoint', () => {
|
||||
new Socket('wss://api-explore.lux.network/socket/v2').connect();
|
||||
expect(lastEs().url).toBe('https://api-explore.lux.network/v1/base/realtime');
|
||||
});
|
||||
|
||||
it('rewrites a bare ws host (test harness shape)', () => {
|
||||
new Socket('ws://localhost:3200').connect();
|
||||
expect(lastEs().url).toBe('http://localhost:3200/v1/base/realtime');
|
||||
});
|
||||
|
||||
it('passes withCredentials through to EventSource', () => {
|
||||
new Socket('wss://api-explore.lux.network', { withCredentials: true }).connect();
|
||||
expect(lastEs().withCredentials).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
describe('SSE Socket — envelope routing', () => {
|
||||
it('maps the backend "blocks" broadcast onto blocks:new_block / new_block', () => {
|
||||
const socket = new Socket('wss://api-explore.lux.network');
|
||||
socket.connect();
|
||||
const channel = socket.channel('blocks:new_block');
|
||||
const handler = vi.fn();
|
||||
channel.on('new_block', handler);
|
||||
|
||||
lastEs().emitOpen();
|
||||
lastEs().emitMessage({ event: 'blocks', chain: 'cchain', data: { number: 42, hash: '0xabc' } });
|
||||
|
||||
expect(handler).toHaveBeenCalledTimes(1);
|
||||
expect(handler).toHaveBeenCalledWith({ block: { number: 42, hash: '0xabc' } });
|
||||
});
|
||||
|
||||
it('routes a topic-scoped envelope straight to that channel + event', () => {
|
||||
const socket = new Socket('wss://api-explore.lux.network');
|
||||
socket.connect();
|
||||
const channel = socket.channel('addresses:0xabc');
|
||||
const handler = vi.fn();
|
||||
channel.on('token_transfer', handler);
|
||||
|
||||
lastEs().emitOpen();
|
||||
const payload = { token_transfers: [ { tx_hash: '0xdead' } ] };
|
||||
lastEs().emitMessage({ event: 'token_transfer', topic: 'addresses:0xabc', data: payload });
|
||||
|
||||
expect(handler).toHaveBeenCalledTimes(1);
|
||||
expect(handler).toHaveBeenCalledWith(payload);
|
||||
});
|
||||
|
||||
it('drops events with no matching channel or translation without throwing', () => {
|
||||
const socket = new Socket('wss://api-explore.lux.network');
|
||||
socket.connect();
|
||||
lastEs().emitOpen();
|
||||
expect(() => lastEs().emitMessage({ event: 'nonexistent', data: {} })).not.toThrow();
|
||||
});
|
||||
|
||||
it('stops delivering to a handler after off()', () => {
|
||||
const socket = new Socket('wss://api-explore.lux.network');
|
||||
socket.connect();
|
||||
const channel = socket.channel('blocks:new_block');
|
||||
const handler = vi.fn();
|
||||
const ref = channel.on('new_block', handler);
|
||||
lastEs().emitOpen();
|
||||
|
||||
channel.off('new_block', ref);
|
||||
lastEs().emitMessage({ event: 'blocks', data: { number: 1 } });
|
||||
|
||||
expect(handler).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
describe('SSE Socket — channel join', () => {
|
||||
it('resolves join "ok" once the stream opens', () => {
|
||||
const socket = new Socket('wss://api-explore.lux.network');
|
||||
socket.connect();
|
||||
const channel = socket.channel('blocks:new_block');
|
||||
const onOk = vi.fn();
|
||||
|
||||
channel.join().receive('ok', onOk);
|
||||
expect(onOk).not.toHaveBeenCalled();
|
||||
|
||||
lastEs().emitOpen();
|
||||
expect(onOk).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('resolves join "ok" immediately when the stream is already open', async() => {
|
||||
const socket = new Socket('wss://api-explore.lux.network');
|
||||
socket.connect();
|
||||
lastEs().emitOpen();
|
||||
|
||||
const channel = socket.channel('blocks:new_block');
|
||||
const onOk = vi.fn();
|
||||
channel.join().receive('ok', onOk);
|
||||
|
||||
await Promise.resolve();
|
||||
expect(onOk).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
});
|
||||
|
||||
describe('SSE Socket — connection state', () => {
|
||||
it('fires onError handlers when the stream errors', () => {
|
||||
const socket = new Socket('wss://api-explore.lux.network');
|
||||
socket.connect();
|
||||
const onError = vi.fn();
|
||||
socket.onError(onError);
|
||||
|
||||
lastEs().emitError();
|
||||
expect(onError).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('stops firing a state handler after off()', () => {
|
||||
const socket = new Socket('wss://api-explore.lux.network');
|
||||
socket.connect();
|
||||
const onError = vi.fn();
|
||||
const ref = socket.onError(onError);
|
||||
|
||||
socket.off([ ref ]);
|
||||
lastEs().emitError();
|
||||
expect(onError).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('closes the EventSource on disconnect', () => {
|
||||
const socket = new Socket('wss://api-explore.lux.network');
|
||||
socket.connect();
|
||||
const es = lastEs();
|
||||
socket.disconnect();
|
||||
expect(es.readyState).toBe(es.CLOSED);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,324 @@
|
||||
// EventSource-backed, Phoenix-Channels-compatible realtime transport.
|
||||
//
|
||||
// The Lux explorer Go backend (luxfi/explorer) serves realtime over
|
||||
// Server-Sent Events at `<socketEndpoint>/v1/base/realtime` — a single
|
||||
// multiplexed stream that wraps every hub broadcast in a JSON envelope:
|
||||
//
|
||||
// event: CONNECT (once, on open)
|
||||
// data: {}
|
||||
//
|
||||
// data: {"event":"<hubChannel>","chain":"<slug>","data":<payload>}
|
||||
//
|
||||
// : ping (keep-alive every 30s)
|
||||
//
|
||||
// There is NO Phoenix WebSocket endpoint (the legacy FE opened
|
||||
// `wss://.../socket/v2/websocket` and got a 404, which surfaced as
|
||||
// "Live updates temporarily delayed"). Rather than rewrite all 27
|
||||
// realtime consumers, this module reimplements the slice of the Phoenix
|
||||
// `Socket`/`Channel` API those consumers use, backed by one EventSource.
|
||||
// The seam (`SocketProvider` / `useSocketChannel` / `useSocketMessage`)
|
||||
// stays; only the transport underneath changes.
|
||||
|
||||
export type MessageRef = string;
|
||||
|
||||
type Payload = unknown;
|
||||
// The transport hands consumers the raw wire payload; each consumer re-types
|
||||
// it for its specific message (see useSocketMessage). This single `any`
|
||||
// boundary mirrors the Phoenix `Channel.on` signature this replaces.
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
type WirePayload = any;
|
||||
type EventHandler = (payload?: WirePayload) => void;
|
||||
type StateHandler = () => void;
|
||||
type StateKind = 'open' | 'close' | 'error';
|
||||
type PushStatus = 'ok' | 'error' | 'timeout';
|
||||
|
||||
// Options accepted by SocketProvider. Phoenix took a large connect-option
|
||||
// bag; EventSource only meaningfully honours credentialed requests.
|
||||
export interface SocketOptions {
|
||||
withCredentials?: boolean;
|
||||
}
|
||||
|
||||
// The multiplexed SSE envelope the backend emits. `topic` is optional:
|
||||
// - present -> the payload is delivered straight to that Phoenix topic,
|
||||
// firing `event` handlers (used by chain/address/token-scoped pushes
|
||||
// and by the component-test harness).
|
||||
// - absent -> the backend hub channel in `event` is mapped to a
|
||||
// Blockscout-style (topic, event) pair via TRANSLATIONS below.
|
||||
interface Envelope {
|
||||
event: string;
|
||||
chain?: string;
|
||||
data?: Payload;
|
||||
topic?: string;
|
||||
}
|
||||
|
||||
interface Translation {
|
||||
topic: string;
|
||||
event: string;
|
||||
transform?: (data: Payload, chain?: string) => Payload;
|
||||
}
|
||||
|
||||
// The single place that maps the backend's hub-channel broadcasts to the
|
||||
// Blockscout-style Phoenix (topic, event) pairs the UI subscribes to.
|
||||
// Today the indexer broadcasts only "blocks" (luxfi/indexer
|
||||
// evm/indexer.go `BroadcastBlock`). Extend here as the backend grows new
|
||||
// GLOBAL broadcast types; SCOPED events (addresses:*, tokens:*, …) arrive
|
||||
// with an explicit `topic` and need no entry.
|
||||
const TRANSLATIONS: Record<string, Array<Translation>> = {
|
||||
blocks: [
|
||||
{ topic: 'blocks:new_block', event: 'new_block', transform: (block) => ({ block }) },
|
||||
],
|
||||
};
|
||||
|
||||
// Normalize any realtime URL the app hands us into the backend's SSE
|
||||
// endpoint. Callers pass a Phoenix-style URL (getSocketUrl appends
|
||||
// `/socket/v2`, the multichain-stats API appends `/socket`, the test
|
||||
// harness passes `ws://host:port`); all collapse to one shape here so the
|
||||
// transport has exactly one way to reach the server.
|
||||
function toSSEUrl(raw: string): string {
|
||||
let u = raw.replace(/^wss:/i, 'https:').replace(/^ws:/i, 'http:');
|
||||
u = u.replace(/\/+$/, '');
|
||||
u = u.replace(/\/socket(?:\/v\d+)?(?:\/websocket)?$/i, '');
|
||||
return `${ u }/v1/base/realtime`;
|
||||
}
|
||||
|
||||
// Push mirrors the chainable `channel.join().receive('ok', …).receive('error', …)`
|
||||
// shape. SSE has no per-topic handshake, so a join resolves 'ok' as soon
|
||||
// as the stream is open; it never resolves 'error' (transport errors are
|
||||
// surfaced at the socket level instead).
|
||||
class Push {
|
||||
private handlers: Partial<Record<PushStatus, Array<(resp?: Payload) => void>>> = {};
|
||||
private status: PushStatus | undefined;
|
||||
private response: Payload;
|
||||
|
||||
receive(status: PushStatus, callback: (resp?: Payload) => void): this {
|
||||
if (this.status === status) {
|
||||
callback(this.response);
|
||||
return this;
|
||||
}
|
||||
const list = this.handlers[status] ?? [];
|
||||
list.push(callback);
|
||||
this.handlers[status] = list;
|
||||
return this;
|
||||
}
|
||||
|
||||
resolve(status: PushStatus, response?: Payload) {
|
||||
if (this.status !== undefined) {
|
||||
return;
|
||||
}
|
||||
this.status = status;
|
||||
this.response = response;
|
||||
this.handlers[status]?.forEach((cb) => cb(response));
|
||||
}
|
||||
}
|
||||
|
||||
export class Channel {
|
||||
readonly topic: string;
|
||||
private socket: Socket;
|
||||
private handlers = new Map<string, Map<number, EventHandler>>();
|
||||
private refSeq = 0;
|
||||
private joinPush: Push | undefined;
|
||||
private joined = false;
|
||||
|
||||
constructor(topic: string, socket: Socket) {
|
||||
this.topic = topic;
|
||||
this.socket = socket;
|
||||
}
|
||||
|
||||
join(): Push {
|
||||
if (!this.joinPush) {
|
||||
this.joinPush = new Push();
|
||||
this.joined = false;
|
||||
}
|
||||
if (this.socket.isOpen()) {
|
||||
// Already streaming — resolve after the synchronous `.receive(...)`
|
||||
// chain has registered its callbacks.
|
||||
queueMicrotask(() => this.resolveJoin());
|
||||
}
|
||||
// Otherwise Socket fires resolveJoin() via notifyOpen() on open.
|
||||
return this.joinPush;
|
||||
}
|
||||
|
||||
leave(): Push {
|
||||
this.handlers.clear();
|
||||
this.joinPush = undefined;
|
||||
this.joined = false;
|
||||
this.socket.removeChannel(this.topic);
|
||||
const push = new Push();
|
||||
queueMicrotask(() => push.resolve('ok', {}));
|
||||
return push;
|
||||
}
|
||||
|
||||
on(event: string, callback: EventHandler): number {
|
||||
const ref = ++this.refSeq;
|
||||
let map = this.handlers.get(event);
|
||||
if (!map) {
|
||||
map = new Map();
|
||||
this.handlers.set(event, map);
|
||||
}
|
||||
map.set(ref, callback);
|
||||
return ref;
|
||||
}
|
||||
|
||||
off(event: string, ref?: number) {
|
||||
const map = this.handlers.get(event);
|
||||
if (!map) {
|
||||
return;
|
||||
}
|
||||
if (ref === undefined) {
|
||||
this.handlers.delete(event);
|
||||
return;
|
||||
}
|
||||
map.delete(ref);
|
||||
if (map.size === 0) {
|
||||
this.handlers.delete(event);
|
||||
}
|
||||
}
|
||||
|
||||
// Internal — deliver an inbound event to every handler registered for it.
|
||||
receive(event: string, payload?: Payload) {
|
||||
this.handlers.get(event)?.forEach((cb) => cb(payload));
|
||||
}
|
||||
|
||||
// Internal — the stream (re)opened; complete a pending join exactly once.
|
||||
notifyOpen() {
|
||||
this.resolveJoin();
|
||||
}
|
||||
|
||||
private resolveJoin() {
|
||||
if (this.joined || !this.joinPush) {
|
||||
return;
|
||||
}
|
||||
this.joined = true;
|
||||
this.joinPush.resolve('ok', {});
|
||||
}
|
||||
}
|
||||
|
||||
export class Socket {
|
||||
private url: string;
|
||||
private withCredentials: boolean;
|
||||
private es: EventSource | undefined;
|
||||
private channels = new Map<string, Channel>();
|
||||
private stateHandlers = new Map<MessageRef, { kind: StateKind; cb: StateHandler }>();
|
||||
private refSeq = 0;
|
||||
|
||||
constructor(url: string, options?: SocketOptions) {
|
||||
this.url = toSSEUrl(url);
|
||||
this.withCredentials = options?.withCredentials ?? false;
|
||||
}
|
||||
|
||||
connect() {
|
||||
if (this.es || typeof EventSource === 'undefined') {
|
||||
return;
|
||||
}
|
||||
const es = new EventSource(this.url, { withCredentials: this.withCredentials });
|
||||
this.es = es;
|
||||
|
||||
const handleOpen = () => {
|
||||
this.fire('open');
|
||||
this.channels.forEach((ch) => ch.notifyOpen());
|
||||
};
|
||||
|
||||
es.onopen = handleOpen;
|
||||
// The backend sends an initial typed `CONNECT` frame; treat it as an
|
||||
// open acknowledgement too (covers proxies that swallow onopen).
|
||||
es.addEventListener('CONNECT', handleOpen);
|
||||
es.onmessage = (event: MessageEvent) => this.dispatch(event.data);
|
||||
es.onerror = () => {
|
||||
// EventSource auto-reconnects while readyState === CONNECTING. Surface
|
||||
// the error to consumers (matches Phoenix onError/onClose semantics);
|
||||
// a healthy endpoint never reaches here under normal operation, so the
|
||||
// "Live updates" alert stays cleared.
|
||||
this.fire('error');
|
||||
if (es.readyState === es.CLOSED) {
|
||||
this.fire('close');
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
disconnect(callback?: () => void) {
|
||||
if (this.es) {
|
||||
this.es.close();
|
||||
this.es = undefined;
|
||||
}
|
||||
this.channels.clear();
|
||||
callback?.();
|
||||
}
|
||||
|
||||
channel(topic: string): Channel {
|
||||
let ch = this.channels.get(topic);
|
||||
if (!ch) {
|
||||
ch = new Channel(topic, this);
|
||||
this.channels.set(topic, ch);
|
||||
}
|
||||
return ch;
|
||||
}
|
||||
|
||||
onOpen(callback: StateHandler): MessageRef {
|
||||
return this.addState('open', callback);
|
||||
}
|
||||
|
||||
onClose(callback: StateHandler): MessageRef {
|
||||
return this.addState('close', callback);
|
||||
}
|
||||
|
||||
onError(callback: StateHandler): MessageRef {
|
||||
return this.addState('error', callback);
|
||||
}
|
||||
|
||||
off(refs: Array<MessageRef>) {
|
||||
refs.forEach((ref) => this.stateHandlers.delete(ref));
|
||||
}
|
||||
|
||||
// Internal — true once the underlying stream is open.
|
||||
isOpen() {
|
||||
return this.es !== undefined && this.es.readyState === this.es.OPEN;
|
||||
}
|
||||
|
||||
// Internal — drop a channel once its last subscriber leaves.
|
||||
removeChannel(topic: string) {
|
||||
this.channels.delete(topic);
|
||||
}
|
||||
|
||||
private addState(kind: StateKind, cb: StateHandler): MessageRef {
|
||||
const ref = String(++this.refSeq);
|
||||
this.stateHandlers.set(ref, { kind, cb });
|
||||
return ref;
|
||||
}
|
||||
|
||||
private fire(kind: StateKind) {
|
||||
this.stateHandlers.forEach((handler) => {
|
||||
if (handler.kind === kind) {
|
||||
handler.cb();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
private dispatch(raw: string) {
|
||||
let envelope: Envelope;
|
||||
try {
|
||||
envelope = JSON.parse(raw) as Envelope;
|
||||
} catch {
|
||||
return;
|
||||
}
|
||||
if (!envelope || typeof envelope.event !== 'string') {
|
||||
return;
|
||||
}
|
||||
|
||||
if (envelope.topic) {
|
||||
this.channels.get(envelope.topic)?.receive(envelope.event, envelope.data);
|
||||
return;
|
||||
}
|
||||
|
||||
const routes = TRANSLATIONS[envelope.event];
|
||||
if (!routes) {
|
||||
return;
|
||||
}
|
||||
for (const route of routes) {
|
||||
const ch = this.channels.get(route.topic);
|
||||
if (!ch) {
|
||||
continue;
|
||||
}
|
||||
ch.receive(route.event, route.transform ? route.transform(envelope.data, envelope.chain) : envelope.data);
|
||||
}
|
||||
}
|
||||
}
|
||||
+2
-2
@@ -1,5 +1,3 @@
|
||||
import type { Channel } from 'phoenix';
|
||||
|
||||
import type * as multichain from '@luxfi/multichain-aggregator-types';
|
||||
import type * as zetaChainCCTXType from '@luxfi/zetachain-cctx-types';
|
||||
import type { AddressCoinBalanceHistoryItem, AddressTokensBalancesSocketMessage } from 'types/api/address';
|
||||
@@ -12,6 +10,8 @@ import type { TokenTransfer } from 'types/api/tokenTransfer';
|
||||
import type { Transaction } from 'types/api/transaction';
|
||||
import type { NewZkEvmBatchSocketResponse } from 'types/api/zkEvmL2';
|
||||
|
||||
import type { Channel } from './sse';
|
||||
|
||||
export type SocketMessageParams = SocketMessage.NewBlock |
|
||||
SocketMessage.NewBlockMultichain |
|
||||
SocketMessage.BlocksIndexStatus |
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import type { Channel } from 'phoenix';
|
||||
import { useEffect, useRef, useState } from 'react';
|
||||
|
||||
import { useSocket } from './context';
|
||||
import type { Channel } from './sse';
|
||||
|
||||
interface Params {
|
||||
topic: string | undefined;
|
||||
|
||||
@@ -1,6 +1,5 @@
|
||||
import type { TestFixture, Page } from '@playwright/test';
|
||||
import type { WebSocket } from 'ws';
|
||||
import { WebSocketServer } from 'ws';
|
||||
import http from 'node:http';
|
||||
|
||||
import type { AddressCoinBalanceHistoryItem, AddressTokensBalancesSocketMessage } from 'types/api/address';
|
||||
import type { NewBlockSocketResponse } from 'types/api/block';
|
||||
@@ -11,7 +10,25 @@ import type { Transaction } from 'types/api/transaction';
|
||||
|
||||
import { port as socketPort } from '../utils/socket';
|
||||
|
||||
export type CreateSocketFixture = () => Promise<WebSocket>;
|
||||
// Component-test realtime harness. The app speaks Server-Sent Events to the
|
||||
// Lux explorer backend's multiplexed `/v1/base/realtime` endpoint (see
|
||||
// lib/socket/sse.ts), so this fixture is a minimal SSE server rather than a
|
||||
// Phoenix WebSocket mock. `createSocket` / `joinChannel` / `sendMessage`
|
||||
// keep their signatures so the consumer specs are unchanged; `sendMessage`
|
||||
// emits the backend's `{ event, topic, data }` envelope, addressing a
|
||||
// channel directly via the optional `topic` field the adapter routes off.
|
||||
|
||||
// SSEStream is the server side of one open EventSource stream. Tests
|
||||
// push frames into it with sendMessage().
|
||||
export class SSEStream {
|
||||
constructor(private readonly res: http.ServerResponse) {}
|
||||
|
||||
send(frame: string) {
|
||||
this.res.write(frame);
|
||||
}
|
||||
}
|
||||
|
||||
export type CreateSocketFixture = () => Promise<SSEStream>;
|
||||
|
||||
type Channel = [string, string, string];
|
||||
|
||||
@@ -21,64 +38,64 @@ export interface SocketServerFixture {
|
||||
|
||||
// eslint-disable-next-line @typescript-eslint/no-unused-vars
|
||||
export const createSocket: TestFixture<CreateSocketFixture, { page: Page }> = async({ page }, use) => {
|
||||
const socketServer = new WebSocketServer({ port: socketPort });
|
||||
|
||||
const connectionPromise = new Promise<WebSocket>((resolve) => {
|
||||
socketServer.on('connection', (socket: WebSocket) => {
|
||||
resolve(socket);
|
||||
});
|
||||
const openResponses = new Set<http.ServerResponse>();
|
||||
let resolveConnection: (conn: SSEStream) => void;
|
||||
const connectionPromise = new Promise<SSEStream>((resolve) => {
|
||||
resolveConnection = resolve;
|
||||
});
|
||||
|
||||
const server = http.createServer((req, res) => {
|
||||
if (req.method === 'HEAD') {
|
||||
res.writeHead(200, { 'Content-Type': 'text/event-stream', 'Access-Control-Allow-Origin': '*' });
|
||||
res.end();
|
||||
return;
|
||||
}
|
||||
|
||||
res.writeHead(200, {
|
||||
'Content-Type': 'text/event-stream',
|
||||
'Cache-Control': 'no-cache',
|
||||
Connection: 'keep-alive',
|
||||
'Access-Control-Allow-Origin': '*',
|
||||
});
|
||||
res.write('event: CONNECT\ndata: {}\n\n');
|
||||
|
||||
openResponses.add(res);
|
||||
res.on('close', () => openResponses.delete(res));
|
||||
resolveConnection(new SSEStream(res));
|
||||
});
|
||||
|
||||
await new Promise<void>((resolve) => server.listen(socketPort, resolve));
|
||||
|
||||
await use(() => connectionPromise);
|
||||
|
||||
socketServer.close();
|
||||
openResponses.forEach((res) => res.end());
|
||||
await new Promise<void>((resolve) => server.close(() => resolve()));
|
||||
};
|
||||
|
||||
export const joinChannel = async(socket: WebSocket, channelName: string) => {
|
||||
return new Promise<[string, string, string]>((resolve, reject) => {
|
||||
socket.on('message', (msg) => {
|
||||
try {
|
||||
const payload = JSON.parse(msg.toString()) as Array<string>;
|
||||
|
||||
if (channelName === payload[2] && payload[3] === 'phx_join') {
|
||||
socket.send(JSON.stringify([
|
||||
payload[0],
|
||||
payload[1],
|
||||
payload[2],
|
||||
'phx_reply',
|
||||
{ response: {}, status: 'ok' },
|
||||
]));
|
||||
|
||||
resolve([ payload[0], payload[1], payload[2] ]);
|
||||
}
|
||||
} catch (error) {
|
||||
reject(error);
|
||||
}
|
||||
});
|
||||
});
|
||||
// SSE is a single read-only multiplexed stream — there is no per-topic join
|
||||
// handshake. `socket` is kept for call-site compatibility; the returned token
|
||||
// is what sendMessage() addresses.
|
||||
export const joinChannel = (socket: SSEStream, channelName: string): Promise<Channel> => {
|
||||
return Promise.resolve([ '', '', channelName ]);
|
||||
};
|
||||
|
||||
export function sendMessage(socket: WebSocket, channel: Channel, msg: 'coin_balance', payload: { coin_balance: AddressCoinBalanceHistoryItem }): void;
|
||||
export function sendMessage(socket: WebSocket, channel: Channel, msg: 'token_balance', payload: { block_number: number }): void;
|
||||
export function sendMessage(socket: WebSocket, channel: Channel, msg: `updated_token_balances_${ string }`, payload: AddressTokensBalancesSocketMessage): void;
|
||||
export function sendMessage(socket: WebSocket, channel: Channel, msg: 'transaction', payload: { transaction: number }): void;
|
||||
export function sendMessage(socket: WebSocket, channel: Channel, msg: 'transaction', payload: { transactions: Array<Transaction> }): void;
|
||||
export function sendMessage(socket: WebSocket, channel: Channel, msg: 'pending_transaction', payload: { pending_transaction: number }): void;
|
||||
export function sendMessage(socket: WebSocket, channel: Channel, msg: 'pending_transaction', payload: { pending_transactions: Array<Transaction> }): void;
|
||||
export function sendMessage(socket: WebSocket, channel: Channel, msg: 'new_block', payload: NewBlockSocketResponse): void;
|
||||
export function sendMessage(socket: WebSocket, channel: Channel, msg: 'verification_result', payload: SmartContractVerificationResponse): void;
|
||||
export function sendMessage(socket: WebSocket, channel: Channel, msg: 'total_supply', payload: { total_supply: number }): void;
|
||||
export function sendMessage(socket: WebSocket, channel: Channel, msg: 'changed_bytecode', payload: Record<string, never>): void;
|
||||
export function sendMessage(socket: WebSocket, channel: Channel, msg: 'fetched_bytecode', payload: { fetched_bytecode: string }): void;
|
||||
export function sendMessage(socket: WebSocket, channel: Channel, msg: 'eth_bytecode_db_lookup_started', payload: Record<string, never>): void;
|
||||
export function sendMessage(socket: WebSocket, channel: Channel, msg: 'smart_contract_was_verified', payload: Record<string, never>): void;
|
||||
export function sendMessage(socket: WebSocket, channel: Channel, msg: 'smart_contract_was_not_verified', payload: Record<string, never>): void;
|
||||
export function sendMessage(socket: WebSocket, channel: Channel, msg: 'token_transfer', payload: { token_transfers: Array<TokenTransfer> }): void;
|
||||
export function sendMessage(socket: WebSocket, channel: Channel, msg: 'fetched_token_instance_metadata', payload: TokenInstanceMetadataSocketMessage): void;
|
||||
export function sendMessage(socket: WebSocket, channel: Channel, msg: string, payload: unknown): void {
|
||||
socket.send(JSON.stringify([
|
||||
...channel,
|
||||
msg,
|
||||
payload,
|
||||
]));
|
||||
export function sendMessage(socket: SSEStream, channel: Channel, msg: 'coin_balance', payload: { coin_balance: AddressCoinBalanceHistoryItem }): void;
|
||||
export function sendMessage(socket: SSEStream, channel: Channel, msg: 'token_balance', payload: { block_number: number }): void;
|
||||
export function sendMessage(socket: SSEStream, channel: Channel, msg: `updated_token_balances_${ string }`, payload: AddressTokensBalancesSocketMessage): void;
|
||||
export function sendMessage(socket: SSEStream, channel: Channel, msg: 'transaction', payload: { transaction: number }): void;
|
||||
export function sendMessage(socket: SSEStream, channel: Channel, msg: 'transaction', payload: { transactions: Array<Transaction> }): void;
|
||||
export function sendMessage(socket: SSEStream, channel: Channel, msg: 'pending_transaction', payload: { pending_transaction: number }): void;
|
||||
export function sendMessage(socket: SSEStream, channel: Channel, msg: 'pending_transaction', payload: { pending_transactions: Array<Transaction> }): void;
|
||||
export function sendMessage(socket: SSEStream, channel: Channel, msg: 'new_block', payload: NewBlockSocketResponse): void;
|
||||
export function sendMessage(socket: SSEStream, channel: Channel, msg: 'verification_result', payload: SmartContractVerificationResponse): void;
|
||||
export function sendMessage(socket: SSEStream, channel: Channel, msg: 'total_supply', payload: { total_supply: number }): void;
|
||||
export function sendMessage(socket: SSEStream, channel: Channel, msg: 'changed_bytecode', payload: Record<string, never>): void;
|
||||
export function sendMessage(socket: SSEStream, channel: Channel, msg: 'fetched_bytecode', payload: { fetched_bytecode: string }): void;
|
||||
export function sendMessage(socket: SSEStream, channel: Channel, msg: 'eth_bytecode_db_lookup_started', payload: Record<string, never>): void;
|
||||
export function sendMessage(socket: SSEStream, channel: Channel, msg: 'smart_contract_was_verified', payload: Record<string, never>): void;
|
||||
export function sendMessage(socket: SSEStream, channel: Channel, msg: 'smart_contract_was_not_verified', payload: Record<string, never>): void;
|
||||
export function sendMessage(socket: SSEStream, channel: Channel, msg: 'token_transfer', payload: { token_transfers: Array<TokenTransfer> }): void;
|
||||
export function sendMessage(socket: SSEStream, channel: Channel, msg: 'fetched_token_instance_metadata', payload: TokenInstanceMetadataSocketMessage): void;
|
||||
export function sendMessage(socket: SSEStream, channel: Channel, msg: string, payload: unknown): void {
|
||||
socket.send(`data: ${ JSON.stringify({ event: msg, topic: channel[2], data: payload }) }\n\n`);
|
||||
}
|
||||
|
||||
@@ -1,6 +1,5 @@
|
||||
import type { UseQueryResult } from '@tanstack/react-query';
|
||||
import { useRouter } from 'next/router';
|
||||
import type { Channel } from 'phoenix';
|
||||
import React from 'react';
|
||||
|
||||
import type { Address } from 'types/api/address';
|
||||
@@ -11,6 +10,7 @@ import type { ResourceError } from 'lib/api/resources';
|
||||
import useApiQuery from 'lib/api/useApiQuery';
|
||||
import { useMultichainContext } from 'lib/contexts/multichain';
|
||||
import getQueryParamString from 'lib/router/getQueryParamString';
|
||||
import type { Channel } from 'lib/socket/sse';
|
||||
import * as stubs from 'stubs/contract';
|
||||
import RoutedTabs from 'toolkit/components/RoutedTabs/RoutedTabs';
|
||||
import DataFetchAlert from 'ui/shared/DataFetchAlert';
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import type { Channel } from 'phoenix';
|
||||
import { Alert } from '@luxfi/ui/alert';
|
||||
import React from 'react';
|
||||
|
||||
import type { SocketMessage } from 'lib/socket/types';
|
||||
@@ -7,8 +7,8 @@ import type { SmartContract } from 'types/api/contract';
|
||||
|
||||
import { route } from 'nextjs-routes';
|
||||
|
||||
import type { Channel } from 'lib/socket/sse';
|
||||
import useSocketMessage from 'lib/socket/useSocketMessage';
|
||||
import { Alert } from '@luxfi/ui/alert';
|
||||
import { Link } from 'toolkit/next/link';
|
||||
import AddressEntity from 'ui/shared/entities/address/AddressEntity';
|
||||
|
||||
|
||||
@@ -1,4 +1,3 @@
|
||||
import type { Channel } from 'phoenix';
|
||||
import React from 'react';
|
||||
|
||||
import type { Address } from 'types/api/address';
|
||||
@@ -6,6 +5,7 @@ import type { ClusterChainConfig } from 'types/multichain';
|
||||
|
||||
import config from 'configs/app';
|
||||
import useApiQuery from 'lib/api/useApiQuery';
|
||||
import type { Channel } from 'lib/socket/sse';
|
||||
import * as stubs from 'stubs/contract';
|
||||
import { ContentLoader } from 'toolkit/components/loaders/ContentLoader';
|
||||
import ContractDetails from 'ui/address/contract/ContractDetails';
|
||||
|
||||
Reference in New Issue
Block a user