Back to OScope
oscope/src/server.mjs
OScope OScope: open source memecoin terminal for Solana f8ff931·Oct 20, 2026 2 Commits
Code
382 lines (348 loc) · 18.3 KB
Raw
// OScope server: reads the live trade stream, keeps the coins in memory, serves the terminal page and talks to it
// over one websocket. It also passes the raw stream on at /firehose, so your own tools can share its one connection.
import http from 'node:http';
import fs from 'node:fs';
import path from 'node:path';
import zlib from 'node:zlib';
import { WebSocketServer } from 'ws';
import { config, ROOT } from './config.mjs';
import { Store } from './store.mjs';
import { Meta } from './meta.mjs';
import { connectFeed, solPrice } from './feed.mjs';
const PUBLIC = path.join(ROOT, 'public');
const log = (...a) => console.log(new Date().toISOString().slice(11, 19), ...a);
const store = new Store();
const meta = new Meta(store);
store.wantMeta = c => meta.want(c);
const loaded = store.load();
if (loaded) log(`loaded ${loaded} coins from ${config.dataDir}`);
// pictures that were being fetched or had failed when OScope stopped are fetched again when their coin is looked at
for (const c of store.coins.values()) if (c.metaState === 1 || c.metaState === 3) { c.metaState = 0; c.metaTries = 0; }
const sol = solPrice(log);
// ---------- firehose: the raw stream, passed on to local tools ----------
// ws://localhost:<port>/firehose every message
// ws://localhost:<port>/firehose?pools=pump,pump-amm only those pools
const fire = new Set();
function relay(f, raw) {
let text = raw;
for (const cl of fire) {
if (cl.pools && !cl.pools.has(f.pool)) continue;
if (cl.ws.readyState !== 1) continue;
if (cl.ws.bufferedAmount > 8e6) { cl.ws.terminate(); continue; } // a reader that cannot keep up is cut, never the stream
cl.ws.send(text ??= JSON.stringify(f));
}
}
const feed = connectFeed(config.feedUrl, (f, raw) => { store.ingest(f); if (fire.size) relay(f, raw); }, log);
// ---------- websocket clients ----------
const clients = new Set();
const coinSubs = new Map(); // mint -> Set(client)
const walletSubs = new Map(); // wallet -> Set(client)
const MINT_RE = /^[1-9A-HJ-NP-Za-km-z]{32,44}$/;
function send(cl, obj) {
if (cl.ws.readyState !== 1) return;
if (cl.ws.bufferedAmount > 16e6) { cl.ws.terminate(); return; }
cl.ws.send(JSON.stringify(obj));
}
function unsubCoin(cl) {
if (!cl.coin) return;
const s = coinSubs.get(cl.coin);
if (s) { s.delete(cl); if (!s.size) { coinSubs.delete(cl.coin); store.pinned.delete(cl.coin); } }
cl.coin = null;
}
function setWallets(cl, list) {
for (const w of cl.wallets) { const s = walletSubs.get(w); if (s) { s.delete(cl); if (!s.size) walletSubs.delete(w); } }
cl.wallets = new Set(list.filter(w => typeof w === 'string' && MINT_RE.test(w)).slice(0, 1000));
for (const w of cl.wallets) { if (!walletSubs.has(w)) walletSubs.set(w, new Set()); walletSubs.get(w).add(cl); }
}
let tape = [];
let tapeOn = 0;
store.onTrade = (c, tr) => {
if (tapeOn) { if (!c.metaState) meta.want(c); tape.push([tr[0], c.mint, c.symbol, c.image ? `/pic/${c.mint}` : '', tr[1], tr[2], tr[5], tr[6], c.pool, tr[7], c.name]); if (tape.length > 3000) tape = tape.slice(-1500); }
const subs = coinSubs.get(c.mint);
if (subs) for (const cl of subs) cl.trades.push(tr);
const ws = walletSubs.get(tr[6]);
if (ws) {
const item = { w: tr[6], m: c.mint, n: c.name, s: c.symbol, img: c.image ? `/pic/${c.mint}` : '', buy: tr[1], sol: tr[2], tok: tr[3], mc: tr[5], ts: tr[0], sig: tr[7] };
for (const cl of ws) cl.wt.push(item);
}
};
const wss = new WebSocketServer({ noServer: true, maxPayload: 256 * 1024, perMessageDeflate: { threshold: 1024 } });
wss.on('connection', ws => {
const cl = { ws, pulse: false, tape: false, sent: new Map(), cols: '', coin: null, coinVer: 0, trades: [], watch: new Set(), wallets: new Set(), wt: [] };
clients.add(cl);
send(cl, { t: 'hello', sol: sol.usd, feed: feed.connected, stretchAt: config.stretchAt });
ws.on('message', buf => {
let msg;
try { msg = JSON.parse(buf.toString()); } catch { return; }
if (!msg || typeof msg !== 'object') return;
if (msg.t === 'pulse') { cl.pulse = !!msg.on; cl.sent.clear(); cl.cols = ''; }
else if (msg.t === 'tape') { if (!!msg.on !== cl.tape) { cl.tape = !!msg.on; tapeOn += cl.tape ? 1 : -1; } }
else if (msg.t === 'coin') {
unsubCoin(cl);
if (typeof msg.mint === 'string' && MINT_RE.test(msg.mint)) {
cl.coin = msg.mint; cl.coinVer = 0; cl.trades = [];
if (!coinSubs.has(msg.mint)) coinSubs.set(msg.mint, new Set());
coinSubs.get(msg.mint).add(cl);
store.pinned.add(msg.mint);
const c = store.coins.get(msg.mint);
if (c) { c.viewed = Date.now(); meta.urgent(c); }
}
} else if (msg.t === 'watch' && Array.isArray(msg.mints)) {
cl.watch = new Set(msg.mints.filter(m => typeof m === 'string' && MINT_RE.test(m)).slice(0, 300));
} else if (msg.t === 'track' && Array.isArray(msg.wallets)) setWallets(cl, msg.wallets);
});
ws.on('close', () => { unsubCoin(cl); setWallets(cl, []); if (cl.tape) tapeOn--; clients.delete(cl); });
});
// Pulse: every 400 ms each page gets the column order (when it changed) and the rows that changed since its last send.
setInterval(() => {
const cols = store.placeColumns();
const ids = { new: cols.new.map(c => c.mint), stretch: cols.stretch.map(c => c.mint), migrated: cols.migrated.map(c => c.mint) };
const idsKey = JSON.stringify(ids);
const all = [...cols.new, ...cols.stretch, ...cols.migrated];
for (const c of all) if (!c.metaState || (c.metaState === 1 && !c.urged)) { c.urged = true; meta.urgent(c); }
for (const cl of clients) {
if (!cl.pulse || cl.ws.bufferedAmount > 4e6) continue;
const rows = [];
const present = new Set();
for (const c of all) {
present.add(c.mint);
if (cl.sent.get(c.mint) !== c.ver) { rows.push(store.row(c)); cl.sent.set(c.mint, c.ver); }
}
for (const m of cl.sent.keys()) if (!present.has(m)) cl.sent.delete(m);
if (!rows.length && cl.cols === idsKey) continue;
const out = { t: 'p', rows };
if (cl.cols !== idsKey) { out.cols = ids; cl.cols = idsKey; }
send(cl, out);
}
}, 400);
// Coin page: live trades every 250 ms, the coin's numbers when they change. Tracked wallets' trades.
setInterval(() => {
for (const cl of clients) {
if (cl.coin) {
const c = store.coins.get(cl.coin);
if (c && cl.coinVer !== c.ver) { cl.coinVer = c.ver; send(cl, { t: 'c', d: store.detail(c) }); }
if (cl.trades.length) { send(cl, { t: 'trs', m: cl.coin, list: cl.trades.slice(-200) }); cl.trades = []; }
}
if (cl.wt.length) { send(cl, { t: 'wt', list: cl.wt.slice(-100) }); cl.wt = []; }
}
if (tape.length) {
const list = tape.slice(-400);
tape = [];
for (const cl of clients) if (cl.tape && cl.ws.bufferedAmount < 4e6) send(cl, { t: 'tape', list, n: store.stats.trades });
}
}, 250);
// Prices of the coins a page holds (paper positions), once a second.
setInterval(() => {
for (const cl of clients) {
if (!cl.watch.size) continue;
const p = {};
for (const m of cl.watch) {
const c = store.coins.get(m);
if (c && c.price) p[m] = [c.price, c.mc, c.pool, c.rq, c.rt, c.fee, c.name, c.symbol, c.image ? `/pic/${c.mint}` : ''];
}
send(cl, { t: 'px', p });
}
}, 1000);
setInterval(() => {
const st = { t: 'st', sol: sol.usd, feed: feed.connected, lag: store.stats.lastFrame ? Date.now() - store.stats.lastFrame : null, coins: store.coins.size, trades: store.stats.trades, fire: fire.size };
for (const cl of clients) send(cl, st);
}, 3000);
setInterval(() => {
// coins open on a page or held in a paper wallet keep their full detail
const pinned = new Set(coinSubs.keys());
for (const cl of clients) for (const m of cl.watch) pinned.add(m);
store.pinned = pinned;
for (const m of coinSubs.keys()) { const c = store.coins.get(m); if (c) c.viewed = Date.now(); }
const n = store.prune();
if (n > 2000) log(`dropped ${n} quiet coins, ${store.coins.size} kept`);
}, 30000);
setInterval(() => { try { store.save(); } catch (e) { log('save failed:', e.message); } }, 60000).unref();
// ---------- http ----------
const TYPES = { '.html': 'text/html; charset=utf-8', '.js': 'text/javascript; charset=utf-8', '.mjs': 'text/javascript; charset=utf-8', '.css': 'text/css; charset=utf-8', '.svg': 'image/svg+xml', '.png': 'image/png', '.ico': 'image/x-icon', '.woff2': 'font/woff2', '.json': 'application/json', '.txt': 'text/plain; charset=utf-8', '.webmanifest': 'application/manifest+json' };
// The page may load and connect to this server only: no outside scripts, pictures, fonts or connections.
const csp = host => `default-src 'self'; img-src 'self' data: blob:; connect-src 'self' ws://${host} wss://${host}; style-src 'self' 'unsafe-inline'; font-src 'self'; script-src 'self'; object-src 'none'; base-uri 'none'; frame-ancestors 'self'`;
const VENDOR = { '/vendor/lightweight-charts.mjs': path.join(ROOT, 'node_modules/lightweight-charts/dist/lightweight-charts.standalone.production.mjs') };
const fileCache = new Map();
function staticFile(file) {
const st = fs.statSync(file);
const hit = fileCache.get(file);
if (hit && hit.mtime === st.mtimeMs) return hit;
const body = fs.readFileSync(file);
const type = TYPES[path.extname(file)] || 'application/octet-stream';
const text = /^(text|application\/(json|manifest))|svg/.test(type);
const v = { mtime: st.mtimeMs, body, type, etag: `"${st.size.toString(36)}-${Math.floor(st.mtimeMs).toString(36)}"`,
br: text ? zlib.brotliCompressSync(body) : null, gz: text ? zlib.gzipSync(body) : null };
fileCache.set(file, v);
return v;
}
function sendStatic(req, res, file, cache) {
let f;
try { f = staticFile(file); } catch { return notFound(res); }
const h = { 'content-type': f.type, etag: f.etag, 'cache-control': cache, 'x-content-type-options': 'nosniff', vary: 'accept-encoding' };
if (f.type.startsWith('text/html')) {
const host = String(req.headers.host || '').replace(/[^a-zA-Z0-9.:\-\[\]]/g, '') || 'localhost';
h['content-security-policy'] = csp(host);
h['referrer-policy'] = 'no-referrer';
}
if (req.headers['if-none-match'] === f.etag) { res.writeHead(304, h); return res.end(); }
const ae = req.headers['accept-encoding'] || '';
let body = f.body;
if (f.br && /\bbr\b/.test(ae)) { body = f.br; h['content-encoding'] = 'br'; }
else if (f.gz && /\bgzip\b/.test(ae)) { body = f.gz; h['content-encoding'] = 'gzip'; }
res.writeHead(200, h);
res.end(req.method === 'HEAD' ? undefined : body);
}
function json(res, obj, status = 200) {
const body = JSON.stringify(obj);
res.writeHead(status, { 'content-type': 'application/json', 'cache-control': 'no-store', 'x-content-type-options': 'nosniff' });
res.end(body);
}
function notFound(res) { res.writeHead(404, { 'content-type': 'text/plain' }); res.end('not found'); }
const SPA = /^\/(|live|coin\/[1-9A-HJ-NP-Za-km-z]{32,44}|portfolio|trackers|wallet\/[1-9A-HJ-NP-Za-km-z]{32,44}|settings)$/;
async function api(req, res, url) {
const parts = url.pathname.split('/').filter(Boolean); // api, what, id
const what = parts[1], id = parts[2];
if (what === 'status') {
return json(res, { feed: { connected: feed.connected, reconnects: feed.reconnects, url: config.feedUrl }, firehose: config.firehose ? { path: '/firehose', readers: fire.size } : null, sol: sol.usd, coins: store.coins.size, clients: clients.size,
stats: store.stats, uptime: Math.round(process.uptime()), memoryMB: Math.round(process.memoryUsage().rss / 1048576),
held: { ...store.memory(), picturesMB: Math.round(meta.imgBytes / 1048576) } });
}
if (what === 'search') {
const q = (url.searchParams.get('q') || '').trim().slice(0, 60);
if (!q) return json(res, []);
if (MINT_RE.test(q) && store.coins.has(q)) return json(res, [store.row(store.coins.get(q))]);
const lq = q.toLowerCase(), out = [];
for (const c of store.coins.values()) {
if (c.origin !== 'pump') continue;
const s = c.symbol.toLowerCase(), n = c.name.toLowerCase();
const score = s === lq ? 3 : s.startsWith(lq) ? 2 : n.includes(lq) ? 1 : c.mint.startsWith(q) ? 1 : 0;
if (score) out.push([score, c]);
}
out.sort((a, b) => b[0] - a[0] || b[1].mc - a[1].mc);
const top = out.slice(0, 20).map(([, c]) => c);
for (const c of top) if (!c.metaState) meta.urgent(c);
return json(res, top.map(c => store.row(c)));
}
if (what === 'dev' && id && MINT_RE.test(id)) {
const list = (store.devs.get(id) || []).map(e => {
const c = store.coins.get(e.m);
return { m: e.m, n: c?.name || e.n, s: c?.symbol || e.s, t: e.t, mig: c?.migratedAt || e.mig, mc: c?.mc || 0, ath: Math.max(c?.ath || 0, e.ath || 0), img: c?.image ? `/pic/${e.m}` : '', live: !!c };
}).sort((a, b) => b.t - a.t);
return json(res, list);
}
if (what === 'wallet' && id && MINT_RE.test(id)) {
const out = [];
for (const c of store.coins.values()) {
const h = c.holders.get(id);
if (!h) continue;
out.push({ m: c.mint, n: c.name, s: c.symbol, img: c.image ? `/pic/${c.mint}` : '', bal: h.bal, bs: h.bs, ss: h.ss, n_: h.n, last: h.last, mc: c.mc, pnl: h.ss - h.bs + h.bal * c.price, value: h.bal * c.price });
}
out.sort((a, b) => b.last - a.last);
return json(res, out.slice(0, 300));
}
if (!id || !MINT_RE.test(id)) return notFound(res);
const c = store.coins.get(id);
if (!c) return json(res, { error: 'not seen', mint: id }, 404);
c.viewed = Date.now();
if (what === 'coin') { meta.urgent(c); return json(res, store.detail(c)); }
if (what === 'candles') {
const tf = Math.max(1, Math.min(86400, parseInt(url.searchParams.get('tf') || '60', 10) || 60));
return json(res, store.candles(c, tf));
}
if (what === 'trades') {
const limit = Math.max(1, Math.min(2000, parseInt(url.searchParams.get('limit') || '200', 10) || 200));
return json(res, c.trades.slice(-limit).reverse());
}
if (what === 'holders') return json(res, store.holdersOf(c, 100));
return notFound(res);
}
// When OScope listens only on this computer, it answers only requests addressed to this computer. A website
// that points its own name at 127.0.0.1 (DNS rebinding) gets nothing.
const LOCAL_ONLY = ['127.0.0.1', 'localhost', '::1'].includes(config.host);
const LOCAL_NAMES = new Set(['localhost', '127.0.0.1', '[::1]']);
function hostOk(req) {
if (!LOCAL_ONLY) return true;
const name = String(req.headers.host || '').replace(/:\d+$/, '').toLowerCase();
return LOCAL_NAMES.has(name) || name.endsWith('.localhost');
}
const server = http.createServer(async (req, res) => {
try {
if (!hostOk(req)) { res.writeHead(403, { 'content-type': 'text/plain' }); return res.end('OScope answers only on localhost'); }
const url = new URL(req.url, 'http://x');
const p = decodeURIComponent(url.pathname);
if (req.method !== 'GET' && req.method !== 'HEAD') { res.writeHead(405); return res.end(); }
if (p.startsWith('/api/')) return await api(req, res, url);
if (p.startsWith('/pic/')) {
const mint = p.slice(5);
if (!MINT_RE.test(mint)) return notFound(res);
const img = await meta.image(mint);
if (!img) return notFound(res);
res.writeHead(200, { 'content-type': img.type, 'cache-control': 'public, max-age=86400', 'x-content-type-options': 'nosniff',
'content-security-policy': "default-src 'none'; style-src 'unsafe-inline'; sandbox" });
return res.end(img.body);
}
if (VENDOR[p]) return sendStatic(req, res, VENDOR[p], 'public, max-age=86400');
if (SPA.test(p)) return sendStatic(req, res, path.join(PUBLIC, 'index.html'), 'no-cache');
const file = path.join(PUBLIC, path.normalize(p));
if (!file.startsWith(PUBLIC + path.sep)) return notFound(res);
return sendStatic(req, res, file, 'no-cache');
} catch (e) {
if (!res.headersSent) { res.writeHead(500, { 'content-type': 'text/plain' }); res.end('error'); }
log('http error:', e.message);
}
});
const fireWss = new WebSocketServer({ noServer: true, maxPayload: 4096, perMessageDeflate: false });
fireWss.on('connection', (ws, req) => {
const pools = new URL(req.url, 'http://x').searchParams.get('pools');
const cl = { ws, pools: pools ? new Set(pools.split(',').map(p => p.trim()).filter(Boolean)) : null };
fire.add(cl);
ws.on('close', () => fire.delete(cl));
ws.on('message', () => {}); // readers only listen
});
server.on('upgrade', (req, sock, head) => {
const p = req.url.split('?')[0];
if (!hostOk(req)) return sock.destroy();
if (p !== '/ws' && !(p === '/firehose' && config.firehose)) return sock.destroy();
// A web page may open these sockets only if this server served it: other websites you visit cannot read them.
// Tools (no Origin header) may.
const origin = req.headers.origin;
if (origin) {
try { if (new URL(origin).host !== req.headers.host) return sock.destroy(); } catch { return sock.destroy(); }
}
const target = p === '/ws' ? wss : fireWss;
target.handleUpgrade(req, sock, head, ws => target.emit('connection', ws, req));
});
// ---------- start ----------
let port = config.port;
function listen() {
server.listen(port, config.host);
}
server.on('error', err => {
if (err.code === 'EADDRINUSE' && !config.portFixed && port < config.port + 20) {
log(`port ${port} is taken, trying ${port + 1}`);
port++;
return setTimeout(listen, 50);
}
if (err.code === 'EADDRINUSE') { console.error(`\nPort ${port} is already in use. Pick another one: PORT=${port + 1} npm start\n`); process.exit(1); }
console.error(err);
process.exit(1);
});
server.on('listening', () => {
const host = config.host === '0.0.0.0' || config.host === '::' ? 'localhost' : config.host;
const url = `http://${host}:${port}`;
console.log(`\n OScope is running\n\n Open ${url}\n Firehose ${config.firehose ? url.replace('http', 'ws') + '/firehose' : 'off'}\n Data ${config.feedUrl}\n`);
});
listen();
// a line every LOG_EVERY seconds, so you can see the trades coming in
let lastTrades = 0;
if (config.logEvery > 0) {
setInterval(() => {
const n = store.stats.trades - lastTrades;
lastTrades = store.stats.trades;
if (!feed.connected) return log('waiting for the stream…');
log(`live: ${n.toLocaleString('en-US')} trades in the last ${config.logEvery} s (${(n / config.logEvery).toFixed(0)}/s) · ${store.stats.creates} launches and ${store.stats.migrations} migrations seen · ${store.coins.size.toLocaleString('en-US')} coins in memory · ${Math.round(process.memoryUsage().rss / 1048576)} MB`);
}, config.logEvery * 1000).unref();
}
const stop = () => { try { store.save(); log('saved'); } catch {} process.exit(0); };
process.on('SIGINT', stop);
process.on('SIGTERM', stop);