// The firehose on its own: one connection to the stream, passed on to as many local readers as you like.
// OScope itself already does this at ws://localhost:3000/firehose; use this when you want the stream without the page.
//
//   npm run firehose                    -> ws://127.0.0.1:3100
//   readers may add ?pools=pump,pump-amm to get only those pools
//
// Settings: FEED_URL (the stream), FIREHOSE_PORT (3100), FIREHOSE_HOST (127.0.0.1).
// Point OScope at it with FEED_URL=ws://127.0.0.1:3100 to run several things on one stream connection.
import { WebSocketServer } from 'ws';
import { config } from '../src/config.mjs';
import { connectFeed } from '../src/feed.mjs';

const port = Number(process.env.FIREHOSE_PORT) || 3100;
const host = process.env.FIREHOSE_HOST || '127.0.0.1';
const readers = new Set();
let n = 0;

const wss = new WebSocketServer({ port, host, maxPayload: 4096, perMessageDeflate: false });
wss.on('connection', (ws, req) => {
  if (req.headers.origin) return ws.close(1008, 'tools only'); // web pages you visit cannot read it
  const pools = new URL(req.url, 'http://x').searchParams.get('pools');
  const r = { ws, pools: pools ? new Set(pools.split(',')) : null };
  readers.add(r);
  ws.on('close', () => readers.delete(r));
});
wss.on('listening', () => console.log(`\n  Firehose on ws://${host}:${port}\n  Stream    ${config.feedUrl}\n`));
wss.on('error', e => { console.error(e.code === 'EADDRINUSE' ? `Port ${port} is in use: FIREHOSE_PORT=${port + 1} npm run firehose` : e.message); process.exit(1); });

connectFeed(config.feedUrl, (f, raw) => {
  n++;
  let text = raw;
  for (const r of readers) {
    if (r.pools && !r.pools.has(f.pool)) continue;
    if (r.ws.bufferedAmount > 8e6) { r.ws.terminate(); continue; }
    r.ws.send(text ??= JSON.stringify(f));
  }
});

setInterval(() => { console.log(`${new Date().toTimeString().slice(0, 8)} ${n.toLocaleString('en-US')} messages in the last 30 s, ${readers.size} reader(s)`); n = 0; }, 30000).unref();