nodejs-visualiser/server.js

const express = require('express');
const http = require('http');
const WebSocket = require('ws');
const axios = require('axios');
const path = require('path');
const zlib = require('zlib');

const argv = require('yargs').argv;
const app = express();
const server = http.createServer(app);
const wss = new WebSocket.Server({ 
  server,
  perMessageDeflate: true
});

let logCache = new Set();
let pollingInterval = 16; // Starting interval
let consecutiveErrors = 0;
const MAX_POLLING_INTERVAL = 200; // Maximum polling interval in ms
const MIN_POLLING_INTERVAL = 16; // Minimum polling interval in ms

app.use(express.static(path.join(__dirname, 'public')));

wss.on('connection', (ws) => {
  debugLog('Client connected:', ws._socket.remoteAddress);

  ws.on('close', () => {
    debugLog('Client disconnected:', ws._socket.remoteAddress);
  });

  ws.on('error', (error) => {
    console.error('WebSocket error:', error);
  });
});

const debugMode = !!argv.debug;
let eventsHandled = 0;
let eventsSentToClients = 0;

function debugLog(...args) {
  if (debugMode) {
    console.log(...args);
  }
}

function periodicReport() {
  console.log(`Last 10 seconds: Handled ${eventsHandled} events, Sent ${eventsSentToClients} events to clients`);
  eventsHandled = 0;
  eventsSentToClients = 0;
}

setInterval(periodicReport, 10000);

async function pollLogs() {
  debugLog('Entering pollLogs function');
  
  let logs;
  try {
    if (debugMode) {
      debugLog('debugMode is true');
      const fs = require('fs');
      const data = fs.readFileSync('logsexample.out', 'utf8');
      const logLines = data.split('\n');
      const validLogs = [];
      logLines.forEach((line, index) => {
        if (line.trim() !== '') {
          try {
            const unescapedLine = line.replace(/\\"/g, '"');
            const parsedLog = JSON.parse(unescapedLine);
            validLogs.push(parsedLog);
          } catch (parseError) {
            console.error(`Error parsing log line at index ${index}: ${parseError.message}`);
          }
        }
      });
      logs = validLogs;
    } else {
      debugLog('debugMode is false');
      
      const query = 'query=_time:2s relay received';
      debugLog('Sending request to API');
      
      const response = await axios.post('https://vmselect.riff.cc/select/logsql/query', query, {
        headers: {
          'Content-Type': 'application/x-www-form-urlencoded'
        }
      });
      
      debugLog('Received response from API');
      
      // Ensure response.data is a string before calling trim
      const responseData = typeof response.data === 'string' ? response.data.trim() : '';
      if (responseData === '') {
        debugLog('Received empty response from API');
      } else {
        debugLog('Received non-empty response from API');
      }
      
      const validLogs = parseLogLines(responseData);
      logs = validLogs;
    }
    
    debugLog('Fetched logs:', logs);
    
    if (Array.isArray(logs) && logs.length > 0) {
      debugLog('logs is a non-empty array, processing...');
      processLogs(logs);
      eventsHandled += logs.length;
      consecutiveErrors = 0;
      // Keep polling interval at minimum when successful
      pollingInterval = MIN_POLLING_INTERVAL;
    } else {
      debugLog('No logs fetched or logs is not an array');
      // Keep polling interval at minimum when no logs are fetched
      pollingInterval = MIN_POLLING_INTERVAL;
    }
  } catch (error) {
    console.error('Error in pollLogs:', error);
    consecutiveErrors++;
    // Increase polling interval on error, but cap at MAX_POLLING_INTERVAL
    pollingInterval = Math.min(MAX_POLLING_INTERVAL, MIN_POLLING_INTERVAL * Math.pow(2, consecutiveErrors));
  }

  // Schedule the next poll
  setTimeout(pollLogs, pollingInterval);
}

// Start polling
pollLogs();

server.listen(3000, () => {
  console.log('Server is listening on port 3000');
});
function parseLogLines(data) {
  const logLines = data.split('\n');
  const validLogs = [];

  logLines.forEach((line, index) => {
    if (line.trim() !== '') {
      try {
        // Check if the line is already valid JSON
        try {
          const parsedLog = JSON.parse(line);
          validLogs.push(parsedLog);
        } catch {
          // If it's not valid JSON, apply transformations
          let jsonLine = line
            .replace(/\\"/g, '"') // Replace escaped quotes with regular quotes
            .replace(/(\w+)=/g, '"$1":') // Replace '=' with ':' to form valid JSON
            .replace(/,(\s*})/g, '$1') // Remove trailing commas before closing braces
            .replace(/_stream":"{(.+?)}"/g, (match, p1) => `_stream":{"${p1.replace(/=/g, '":"').replace(/,/g, '","')}"}`) // Transform _stream field into a JSON object
            .replace(/_msg":"(.+?)"/g, (match, p1) => `_msg":"${p1.replace(/([a-zA-Z0-9]+)=/g, '$1:').replace(/([a-zA-Z0-9]+): /g, '"$1": ').replace(/, /g, ',').replace(/([a-zA-Z0-9]+):/g, '"$1":')}"`); // Transform _msg field into a JSON object

          const parsedLog = JSON.parse(jsonLine);
          validLogs.push(parsedLog);
        }
      } catch (parseError) {
        debugLog(`Error parsing log line at index ${index}: ${parseError.message}`);
        debugLog(`Log line: ${line}`);
        const position = parseError.message.match(/position (\d+)/);
        if (position) {
          const pos = parseInt(position[1], 10);
          debugLog(`Character at position ${pos}: ${line.charAt(pos)}`);
        }
      }
    }
  });

  return validLogs;
}

let updateBatch = [];
const BATCH_INTERVAL = 1000 / 60; // ~16.67ms for 60 updates per second

function processLogs(logs) {
  logs.forEach((log, index) => {
    debugLog(`Processing log at index ${index}`);
    if (!log._msg) {
      debugLog(`Log at index ${index} is missing required fields:`, log);
      return;
    }

    const msgMatch = log._msg.match(/msg_hash=0x[0-9a-fA-F]+/);
    const timeMatch = log._msg.match(/receivedTime=\d+/);
    const peerIdMatch = log._msg.match(/my_peer_id=([^\s]+)/);

    if (msgMatch && timeMatch && peerIdMatch) {
      const msg_hash = msgMatch[0].split('=')[1];
      const receivedTime = timeMatch[0].split('=')[1];
      const peerId = peerIdMatch[1];

      const logIdentifier = `${msg_hash}-${peerId}`;
      if (!logCache.has(logIdentifier)) {
        logCache.add(logIdentifier);
        const logData = {
          msg_hash,
          receivedTime,
          peerId,
          newNode: !logCache.has(`node-${peerId}`)
        };

        logCache.add(`node-${peerId}`);
        updateBatch.push(logData);
        eventsSentToClients++;
      } else {
        debugLog(`Duplicate log found: ${logIdentifier}`);
      }
    } else {
      debugLog(`Log at index ${index} did not match expected format:`, {
        msgMatch: !!msgMatch,
        timeMatch: !!timeMatch,
        peerIdMatch: !!peerIdMatch,
        log: log
      });
    }
  });
}

function sendBatchedUpdates() {
  if (updateBatch.length > 0) {
    const batchData = JSON.stringify(updateBatch);
    wss.clients.forEach(client => {
      if (client.readyState === WebSocket.OPEN) {
        client.send(batchData);
      }
    });
    updateBatch = [];
  }
}

setInterval(sendBatchedUpdates, BATCH_INTERVAL);