// 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);