log-streamer/server.js

'use strict';

const fs = require('node:fs');
const fsp = require('node:fs/promises');
const http = require('node:http');
const path = require('node:path');
const { URL } = require('node:url');
const WebSocket = require('ws');

const PORT = Number.parseInt(process.env.PORT || '3000', 10);
const HOST = process.env.HOST || '127.0.0.1';
const TAIL_LINES = Number.parseInt(process.env.TAIL_LINES || '200', 10);
const WATCH_INTERVAL_MS = Number.parseInt(process.env.WATCH_INTERVAL_MS || '500', 10);
const SHUTDOWN_TIMEOUT_MS = Number.parseInt(process.env.SHUTDOWN_TIMEOUT_MS || '3000', 10);
const PUBLIC_DIR = path.join(__dirname, 'public');
const LOG_FILE = path.resolve(process.env.LOG_FILE || process.argv[2] || 'sample.log');

let offset = 0;
let changeQueue = Promise.resolve();

const mimeTypes = new Map([
  ['.html', 'text/html; charset=utf-8'],
  ['.css', 'text/css; charset=utf-8'],
  ['.js', 'text/javascript; charset=utf-8'],
  ['.json', 'application/json; charset=utf-8'],
  ['.txt', 'text/plain; charset=utf-8']
]);

async function assertLogFile(filePath) {
  let stat;

  try {
    stat = await fsp.stat(filePath);
  } catch (error) {
    if (error.code === 'ENOENT') {
      throw new Error(`Log file does not exist: ${filePath}`);
    }

    throw error;
  }

  if (!stat.isFile()) {
    throw new Error(`Log path is not a file: ${filePath}`);
  }

  return stat;
}

async function readLastLines(filePath, maxLines) {
  const stat = await fsp.stat(filePath);

  if (stat.size === 0 || maxLines <= 0) {
    return [];
  }

  const chunkSize = 64 * 1024;
  const buffers = [];
  let position = stat.size;
  let newlineCount = 0;

  const file = await fsp.open(filePath, 'r');

  try {
    while (position > 0 && newlineCount <= maxLines) {
      const readSize = Math.min(chunkSize, position);
      position -= readSize;

      const buffer = Buffer.allocUnsafe(readSize);
      await file.read(buffer, 0, readSize, position);
      buffers.unshift(buffer);

      for (let i = 0; i < buffer.length; i += 1) {
        if (buffer[i] === 10) {
          newlineCount += 1;
        }
      }
    }
  } finally {
    await file.close();
  }

  let text = Buffer.concat(buffers).toString('utf8').replace(/\r\n/g, '\n');
  const readFromStart = position === 0;

  if (!readFromStart) {
    const firstNewline = text.indexOf('\n');
    text = firstNewline === -1 ? '' : text.slice(firstNewline + 1);
  }

  if (text.endsWith('\n')) {
    text = text.slice(0, -1);
  }

  if (!text) {
    return [];
  }

  return text.split('\n').slice(-maxLines);
}

function sendJson(socket, payload) {
  if (socket.readyState === WebSocket.OPEN) {
    socket.send(JSON.stringify(payload));
  }
}

function broadcast(server, payload) {
  const body = JSON.stringify(payload);

  for (const client of server.clients) {
    if (client.readyState === WebSocket.OPEN) {
      client.send(body);
    }
  }
}

function resolveStaticPath(requestUrl) {
  const parsed = new URL(requestUrl, `http://localhost:${PORT}`);
  const pathname = decodeURIComponent(parsed.pathname);
  const requestedPath = pathname === '/' ? '/index.html' : pathname;
  const filePath = path.resolve(PUBLIC_DIR, `.${requestedPath}`);

  if (filePath !== PUBLIC_DIR && !filePath.startsWith(`${PUBLIC_DIR}${path.sep}`)) {
    return null;
  }

  return filePath;
}

async function serveStatic(request, response) {
  if (request.method !== 'GET' && request.method !== 'HEAD') {
    response.writeHead(405, { Allow: 'GET, HEAD' });
    response.end('Method not allowed');
    return;
  }

  const filePath = resolveStaticPath(request.url);

  if (!filePath) {
    response.writeHead(400, { 'Content-Type': 'text/plain; charset=utf-8' });
    response.end('Bad request');
    return;
  }

  try {
    const stat = await fsp.stat(filePath);

    if (!stat.isFile()) {
      response.writeHead(404, { 'Content-Type': 'text/plain; charset=utf-8' });
      response.end('Not found');
      return;
    }

    const contentType = mimeTypes.get(path.extname(filePath)) || 'application/octet-stream';
    response.writeHead(200, {
      'Content-Type': contentType,
      'Content-Length': stat.size,
      'Cache-Control': 'no-store'
    });

    if (request.method === 'HEAD') {
      response.end();
      return;
    }

    fs.createReadStream(filePath).pipe(response);
  } catch (error) {
    if (error.code === 'ENOENT') {
      response.writeHead(404, { 'Content-Type': 'text/plain; charset=utf-8' });
      response.end('Not found');
      return;
    }

    console.error(error);
    response.writeHead(500, { 'Content-Type': 'text/plain; charset=utf-8' });
    response.end('Internal server error');
  }
}

async function readAppendedBytes(from, to) {
  if (to <= from) {
    return '';
  }

  const stream = fs.createReadStream(LOG_FILE, {
    encoding: 'utf8',
    start: from,
    end: to - 1
  });

  let text = '';

  for await (const chunk of stream) {
    text += chunk;
  }

  return text;
}

function queueFileChange(wss, current) {
  changeQueue = changeQueue
    .then(async () => {
      if (current.size < offset) {
        offset = current.size;
        broadcast(wss, {
          type: 'truncate',
          message: 'File was truncated or rotated; stream reset.'
        });
        return;
      }

      if (current.size === offset) {
        return;
      }

      const text = await readAppendedBytes(offset, current.size);
      offset = current.size;

      if (text) {
        broadcast(wss, { type: 'append', text });
      }
    })
    .catch((error) => {
      console.error(error);
      broadcast(wss, { type: 'error', message: error.message });
    });
}

async function sendInitialState(socket) {
  const stat = await fsp.stat(LOG_FILE);
  const lines = await readLastLines(LOG_FILE, TAIL_LINES);

  sendJson(socket, {
    type: 'hello',
    file: LOG_FILE,
    size: stat.size,
    tailLines: TAIL_LINES
  });
  sendJson(socket, { type: 'snapshot', lines });
}

async function main() {
  const stat = await assertLogFile(LOG_FILE);
  offset = stat.size;

  const server = http.createServer(serveStatic);
  const wss = new WebSocket.Server({ noServer: true });

  wss.on('connection', (socket) => {
    sendInitialState(socket).catch((error) => {
      sendJson(socket, { type: 'error', message: error.message });
      socket.close(1011, 'Initial read failed');
    });
  });

  server.on('upgrade', (request, socket, head) => {
    const parsed = new URL(request.url, `http://localhost:${PORT}`);

    if (parsed.pathname !== '/stream') {
      socket.write('HTTP/1.1 404 Not Found\r\n\r\n');
      socket.destroy();
      return;
    }

    wss.handleUpgrade(request, socket, head, (websocket) => {
      wss.emit('connection', websocket, request);
    });
  });

  fs.watchFile(LOG_FILE, { interval: WATCH_INTERVAL_MS }, (current) => {
    queueFileChange(wss, current);
  });

  server.listen(PORT, HOST, () => {
    console.log(`Log streamer running at http://${HOST}:${PORT}`);
    console.log(`Streaming ${LOG_FILE}`);
  });

  let shuttingDown = false;

  function shutdown(signal) {
    if (shuttingDown) {
      console.error(`${signal} received again; forcing shutdown.`);
      process.exit(1);
    }

    shuttingDown = true;
    console.log(`${signal} received; shutting down.`);
    fs.unwatchFile(LOG_FILE);

    const shutdownTimer = setTimeout(() => {
      console.error(`Shutdown exceeded ${SHUTDOWN_TIMEOUT_MS}ms; forcing exit.`);
      process.exit(1);
    }, SHUTDOWN_TIMEOUT_MS);
    shutdownTimer.unref();

    const terminateTimer = setTimeout(() => {
      for (const client of wss.clients) {
        if (client.readyState !== WebSocket.CLOSED) {
          client.terminate();
        }
      }
    }, Math.min(500, SHUTDOWN_TIMEOUT_MS));
    terminateTimer.unref();

    for (const client of wss.clients) {
      client.close(1001, 'Server shutting down');
    }

    wss.close();

    server.close((error) => {
      clearTimeout(shutdownTimer);

      if (error) {
        console.error(error);
        process.exit(1);
      }

      process.exit(0);
    });
  }

  process.on('SIGINT', () => shutdown('SIGINT'));
  process.on('SIGTERM', () => shutdown('SIGTERM'));
}

main().catch((error) => {
  console.error(error.message);
  process.exit(1);
});