server.js

import 'dotenv/config';

import http from 'node:http';
import { WebSocketServer } from 'ws';

import { createApp } from './app.js';
import { CONFIG } from './config.js';
import { getIpSecretSnapshot, rotateIpSecret, clearIpSecret, secretRefreshMs } from './ws_secret.js';
import { ipFromReq as ipFromNodeReq } from './ip.js';

let broadcastRoundSolved = (_payload) => {};

const { app } = createApp({
  requirePowSecret: true,
  onRoundSolved: (payload) => {
    try { broadcastRoundSolved(payload); } catch { /* ignore */ }
  },
});

const server = http.createServer(app);

// --- WebSocket: global hashrate stats + single-connection-per-IP ---
// Protocol:
//   Client -> Server:
//     - { type: 'hashrate', hps: number, track?: 'CPU_HASH' | 'MEM_WORK' | 'CHAIN' }
//   Server -> Client:
//     - { type: 'stats', network_hps: number, miners: number }
//     - { type: 'secret', secret: string, rotated_at_ms: number, refresh_ms: number, expires_at_ms: number }
//
// NOTE:
// We aggregate *across all tracks*:
// - network_hps = sum(hps) over (ip, track) that are not stale
// - miners = count(distinct ip) that has any non-stale track rate

/** @type {Map<string, import('ws').WebSocket>} */
const ipToWs = new Map();
/** @type {Map<string, Map<string, { hps: number, lastAt: number }>>} */
const ipToTrackRate = new Map();

function ipFromReq(req) {
  return ipFromNodeReq(req, CONFIG.server.trustProxy);
}

function computeNetworkStats() {
  const now = Date.now();
  let sum = 0;
  let miners = 0;

  for (const [_ip, trackMap] of ipToTrackRate.entries()) {
    if (!trackMap || trackMap.size === 0) continue;

    let ipSum = 0;
    for (const [_track, v] of trackMap.entries()) {
      // drop stale rates (no update for 10s)
      if (!v || now - v.lastAt > 10_000) continue;
      ipSum += v.hps;
    }

    if (ipSum > 0) {
      miners++;
      sum += ipSum;
    }
  }

  return { network_hps: sum, miners };
}

// Throttle stats broadcasts to avoid flooding clients under high hashrate message volume.
// (We still compute rates per message; we just limit send frequency.)
const STATS_BROADCAST_MIN_MS = 300;
let statsBroadcastScheduled = false;
let statsBroadcastLastAt = 0;

function broadcastStatsNow() {
  const s = computeNetworkStats();
  const msg = JSON.stringify({ type: 'stats', ...s });
  for (const ws of ipToWs.values()) {
    if (ws.readyState === ws.OPEN) ws.send(msg);
  }
}

function broadcastStats() {
  const now = Date.now();
  const since = now - statsBroadcastLastAt;

  if (since >= STATS_BROADCAST_MIN_MS) {
    statsBroadcastLastAt = now;
    broadcastStatsNow();
    return;
  }

  if (statsBroadcastScheduled) return;
  statsBroadcastScheduled = true;
  const delay = Math.max(0, STATS_BROADCAST_MIN_MS - since);
  setTimeout(() => {
    statsBroadcastScheduled = false;
    statsBroadcastLastAt = Date.now();
    broadcastStatsNow();
  }, delay).unref?.();
}

function broadcastJson(obj) {
  const msg = JSON.stringify(obj);
  for (const ws of ipToWs.values()) {
    if (ws.readyState === ws.OPEN) ws.send(msg);
  }
}

const wss = new WebSocketServer({ server, path: '/ws' });

broadcastRoundSolved = ({ track, round_id, winner_ip }) => {
  // winner_ip is informational only; do not rely on it for auth
  broadcastJson({ type: 'round_solved', track, round_id, winner_ip });
};

wss.on('connection', (ws, req) => {
  const ip = ipFromReq(req);

  // single-connection-per-IP: kick old
  const prev = ipToWs.get(ip);
  if (prev && prev.readyState === prev.OPEN) {
    try {
      prev.send(JSON.stringify({ type: 'kicked', reason: 'new_connection' }));
    } catch {
      // ignore
    }
    try {
      prev.close(4000, 'replaced_by_new_connection');
    } catch {
      // ignore
    }
  }

  ipToWs.set(ip, ws);
  ipToTrackRate.set(ip, new Map());

  // secret push timer (per connection)
  const sendSecret = () => {
    const snap = getIpSecretSnapshot(ip);
    // Include server-side IP so the client can debug bad_secret issues when behind proxies.
    ws.send(JSON.stringify({ type: 'secret', ip, ...snap }));
  };

  // send immediately + then rotate every REFRESH_MS
  sendSecret();
  const secretTimer = setInterval(() => {
    rotateIpSecret(ip);
    if (ws.readyState === ws.OPEN) sendSecret();
  }, secretRefreshMs());

  ws.on('message', (data) => {
    let msg;
    try {
      msg = JSON.parse(data.toString('utf8'));
    } catch {
      return;
    }
    if (msg?.type === 'hashrate') {
      const hps = Number(msg.hps);
      if (!Number.isFinite(hps) || hps < 0 || hps > 1e12) return;

      // Client may include track; if absent, treat as aggregated "ALL".
      const track = typeof msg.track === 'string' && msg.track.length ? msg.track : 'ALL';

      const m = ipToTrackRate.get(ip) ?? new Map();
      m.set(track, { hps, lastAt: Date.now() });
      ipToTrackRate.set(ip, m);

      broadcastStats();
    }
  });

  ws.on('close', () => {
    clearInterval(secretTimer);
    const cur = ipToWs.get(ip);
    if (cur === ws) {
      ipToWs.delete(ip);
      // Only clear secret if this was the active connection.
      // (If a new connection kicked this one, keep secret state for the new one.)
      clearIpSecret(ip);
    }
    ipToTrackRate.delete(ip);
    broadcastStats();
  });

  ws.on('error', () => {
    // handled by close
  });

  // initial stats
  try {
    const s = computeNetworkStats();
    ws.send(JSON.stringify({ type: 'stats', ...s }));
  } catch {
    // ignore
  }

  broadcastStats();
});

setInterval(broadcastStats, 2_000).unref();

server.listen(CONFIG.server.port, () => {
  console.log(`powloot server listening on :${CONFIG.server.port}`);
});