'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); });