waves/index.mjs

import cluster from 'cluster';
import compression from 'compression';
import express from 'express';
import fs from 'fs';
import { createServer } from 'http';
import { spawnSync } from 'child_process';
import net from 'net';
import os from 'os';
import path from 'path';
import WebSocket from 'ws';
import { LRUCache } from 'lru-cache';
import wisp from 'wisp-server-node';

import { baremuxPath } from '@mercuryworkshop/bare-mux/node';
import { epoxyPath } from '@mercuryworkshop/epoxy-transport';
import { libcurlPath } from '@mercuryworkshop/libcurl-transport';
import { uvPath } from '@titaniumnetwork-dev/ultraviolet';

import './others/scaler.mjs';
import './others/warmup.mjs';

const surgeConfigPath = path.resolve('surge.config.json');
const isSurgedRun = process.argv.includes('--surged');
const PORT = parseInt(process.env.PORT || '3000', 10);
let startTime = Date.now();

applySurgeAndRestartIfNeeded();

if (global.gc) {
  setInterval(() => {
    const { heapUsed, heapTotal } = process.memoryUsage();
    if (heapTotal > 0 && heapUsed / heapTotal > 0.7) global.gc();
  }, 60000);
}

function logInfo(msg) { console.info(`[~] ${msg}`); }
function logSuccess(msg) { console.info(`[+] ${msg}`); }
function logError(err) { console.error(`[!] ${err instanceof Error ? err.message : err}`); }

process.on('uncaughtException', err => logError(`Unhandled Exception: ${err}`));
process.on('unhandledRejection', reason => logError(`Unhandled Rejection: ${reason}`));

function applySurgeAndRestartIfNeeded() {
  if (isSurgedRun) {
    try {
      const config = JSON.parse(fs.readFileSync(surgeConfigPath, 'utf-8'));
      process.env.UV_THREADPOOL_SIZE = String(config.uvThreadpoolSize);
    } catch {}
    return;
  }

  const prep = spawnSync(process.execPath, ['./others/surge.mjs'], { stdio: 'inherit' });
  if (prep.error) process.exit(1);

  const { nodeFlags, uvThreadpoolSize } = JSON.parse(fs.readFileSync(surgeConfigPath, 'utf-8'));
  const args = [...nodeFlags, path.resolve('index.mjs'), '--surged'];
  const env = { ...process.env, UV_THREADPOOL_SIZE: String(uvThreadpoolSize), ALREADY_SURGED: 'true' };

  const relaunch = spawnSync(process.execPath, args, { stdio: 'inherit', env });
  process.exit(relaunch.status || 0);
}

if (cluster.isPrimary) {
  const cpuCount = Math.max(1, os.cpus().length - 1);
  logInfo(`Master: forking ${cpuCount} workers`);

  for (let i = 0; i < cpuCount; i++) {
    cluster.fork();
  }

  cluster.on('exit', worker => {
    logError(`Worker ${worker.process.pid} exited. Restarting...`);
    cluster.fork();
  });

  const balancer = net.createServer({ pauseOnConnect: true }, conn => {
    const workers = Object.values(cluster.workers);
    if (workers.length === 0) return conn.destroy();
    const worker = workers[(balancer.current = (balancer.current + 1 || 0)) % workers.length];
    worker.send('sticky-session:connection', conn);
  });

  balancer.on('error', err => logError(`Balancer error: ${err}`));
  balancer.listen(PORT, () => logSuccess(`Balancer listening on ${PORT}`));
} else {
  const app = express();
  const publicPath = path.join(process.cwd(), 'public');
  const cache = new LRUCache({
    maxSize: 1000,
    ttl: 345600000,
    sizeCalculation: (value, key) => Buffer.byteLength(value) + Buffer.byteLength(key)
  });
  const latencySamples = [];

  app.use(compression({ level: 4, memLevel: 4, threshold: 1024 }));

  app.use((req, res, next) => {
    const key = req.originalUrl;
    const cached = cache.get(key);
    if (cached) {
      res.setHeader('X-Cache', 'HIT');
      return res.send(cached);
    }

    res.sendResponse = res.send.bind(res);
    res.send = body => {
      cache.set(key, body);
      res.setHeader('X-Cache', 'MISS');
      res.sendResponse(body);
    };
    next();
  });

  const staticOpts = { maxAge: '7d', immutable: true };
  app.use('/baremux', express.static(baremuxPath, staticOpts));
  app.use('/epoxy', express.static(epoxyPath, staticOpts));
  app.use('/libcurl', express.static(libcurlPath, staticOpts));
  app.use('/wah', express.static(uvPath, staticOpts));
  app.use(express.static(publicPath, staticOpts));
  app.use(express.json());

  const sendHtml = file => (_req, res) => res.sendFile(path.join(publicPath, file));
  app.get('/', sendHtml('$.html'));
  app.get('/g', sendHtml('!.html'));
  app.get('/a', sendHtml('!!.html'));
  app.get('/resent', (_req, res) => res.sendFile(path.join(publicPath, 'resent/index.html')));

  app.get('/api/info', (_req, res) => {
    try {
      const avg = latencySamples.length
        ? latencySamples.reduce((sum, v) => sum + v, 0) / latencySamples.length
        : 0;
      let speed = 'Medium';
      if (avg < 200) speed = 'Fast';
      else if (avg > 500) speed = 'Slow';

      res.json({
        speed,
        averageLatency: avg.toFixed(2),
        specs: `${os.cpus()[0].model} + ${os.cpus().length} cores + ${(os.totalmem() / 1e9).toFixed(1)}GB RAM`,
        startTime,
        samples: latencySamples.length,
        timestamp: Date.now()
      });
    } catch {
      res.status(500).json({ error: 'Internal error' });
    }
  });

  app.get('/api/github-updates', async (_req, res) => {
    try {
      const ghRes = await fetch(
        'https://api.github.com/repos/xojw/waves/commits?per_page=1',
        {
          headers: {
            'User-Agent': 'waves-app',
            'Accept': 'application/vnd.github.v3+json'
          }
        }
      );
      if (!ghRes.ok) {
        return res.status(ghRes.status).json({ error: 'GitHub API error' });
      }

      const [latest] = await ghRes.json();
      const then = new Date(latest.commit.author.date).getTime();
      const diff = Date.now() - then;
      const mins = Math.round(diff / 60000);
      const hrs = Math.round(diff / 3600000);
      const ago = hrs >= 1
        ? `${hrs} hour${hrs > 1 ? 's' : ''} ago`
        : `${mins} minute${mins !== 1 ? 's' : ''} ago`;

      res.json({
        repo: 'xojw/waves',
        update: {
          sha: latest.sha.slice(0, 7),
          author: latest.commit.author.name,
          message: latest.commit.message.split('\n')[0],
          date: latest.commit.author.date,
          ago
        }
      });
    } catch (err) {
      logError(err);
      res.status(500).json({ error: 'Internal server error' });
    }
  });

  app.use((_req, res) => res.status(404).sendFile(path.join(publicPath, '404.html')));

  const server = createServer(app);
  server.keepAliveTimeout = 0;
  server.headersTimeout = 0;

  const pingWSS = new WebSocket.Server({
    noServer: true,
    maxPayload: 4194304,
    perMessageDeflate: false
  });

  pingWSS.on('connection', (ws, req) => {
    const remote = req.socket.remoteAddress || 'unknown';
    let lat = [];

    const ticker = setInterval(() => {
      if (ws.readyState === WebSocket.OPEN) {
        ws.send(JSON.stringify({ type: 'ping', timestamp: Date.now() }));
      }
    }, 1000);

    ws.on('message', msg => {
      try {
        const { type, timestamp } = JSON.parse(msg);
        if (type === 'pong' && timestamp) {
          const d = Date.now() - timestamp;
          lat.push(d);
          latencySamples.push(d);
          if (lat.length > 5) lat.shift();
        }
      } catch {}
    });

    ws.on('close', () => {
      clearInterval(ticker);
    });
  });

  server.on('upgrade', (req, socket, head) => {
    if (req.url === '/ws/ping') {
      pingWSS.handleUpgrade(req, socket, head, ws => {
        pingWSS.emit('connection', ws, req);
      });
    } else {
      socket.destroy();
    }
  });

  server.listen(0, () => {
    logSuccess(`Worker ${process.pid} ready`);
  });

  process.on('message', (msg, conn) => {
    if (msg !== 'sticky-session:connection') return;
    server.emit('connection', conn);
    conn.resume();
  });
}