search ipaddr from logfile

searchip2.mjs

import fs from 'node:fs';
import path from 'node:path';
import cp from 'child_process';
import { Worker, isMainThread, parentPort } from 'node:worker_threads';
import { performance } from 'node:perf_hooks';
import os from 'node:os';
import { fileURLToPath } from 'node:url';

const __filename = fileURLToPath(import.meta.url);
const numWorkers = Math.min(16,os.cpus().length);
const CHUNK_SIZE = 128 * 1024 * 1024; // 128MB chunks

// Allocate 512MB to map every possible IPv4 address (2^32 bits)
const bitsetSAB = new SharedArrayBuffer(1024 * 1024 * 512);
const bitsetView = new Uint32Array(bitsetSAB);

if (isMainThread) {
    let inputFile = process.argv[2];
    if (!inputFile) {
        console.error(`Usage: node ${path.basename(process.argv[1])} <logfile>\n       Search IPv4 from logfile, check unique_ips.txt for results.`);
        process.exit(1);
    }

    if (!fs.existsSync(inputFile)) {
        console.error(`\n❌ Error: File not found: "${inputFile}"`);
        console.error(`Please check the path and try again.`);
        process.exit(1);
    }
	
    let isTemp = false;
	if (/\.(gz|zst|bz2|zip|br|7z|rar|tar)$/i.test(inputFile)) {
		const decompressedPath = `${inputFile}.tmp`;
		console.log(`🚀 Extracting ${path.basename(inputFile)}...`);

		try {
			const outFd = fs.openSync(decompressedPath, 'w');
			const result = cp.spawnSync('7z', ['e', inputFile, '-so', '-y'], {
				stdio: ['ignore', outFd, 'inherit'] 
			});

			fs.closeSync(outFd);

			if (result.status !== 0) {
				throw new Error(`7z exited with code ${result.status}`);
			}

			inputFile = decompressedPath;
			isTemp = true;
		} catch (err) {
			console.error("❌ Decompression failed:", err.message);
			if (fs.existsSync(decompressedPath)) fs.unlinkSync(decompressedPath);
			process.exit(1);
		}
	}


	
    const stats = fs.statSync(inputFile);
    const fd = fs.openSync(inputFile, 'r');
    const startTime = performance.now();
    let filePos = 0;
    let activeWorkers = 0;
    let uiInterval;

    console.log(`\n💎 SCAN | ${numWorkers} Threads | 512MB Shared Bitset`);

    // Master Message Handler
    const handleMessage = (worker) => (msg) => {
        if (filePos < stats.size) {
            const sab = new SharedArrayBuffer(CHUNK_SIZE);
            const bytesRead = fs.readSync(fd, new Uint8Array(sab), 0, CHUNK_SIZE, filePos);
            
            // Handle split IPs: back up to the last non-IP character found by worker
            if (msg.lastMatchEnd !== undefined && msg.lastMatchEnd !== -1) {
                const backtrack = CHUNK_SIZE - msg.lastMatchEnd;
                filePos -= backtrack;
            }

            filePos += bytesRead;
            worker.postMessage({ sab, length: bytesRead });
        } else {
            worker.terminate();
            if (--activeWorkers === 0) finalize();
        }
    };

    // Spawn Workers
    for (let i = 0; i < numWorkers; i++) {
        const worker = new Worker(__filename);
        activeWorkers++;
        worker.on('message', handleMessage(worker));
        worker.on('error', (err) => console.error("Worker Error:", err));
        worker.postMessage({ type: 'start', bitsetSAB });
    }

    // Progress UI
    uiInterval = setInterval(() => {
        const elapsed = (performance.now() - startTime) / 1000;
        const speed = (filePos / 1024 / 1024 / elapsed).toFixed(1);
        const prog = ((filePos / stats.size) * 100).toFixed(1);
        process.stdout.write(`\r⚡ Speed: ${speed} MB/s | Progress: ${prog}% | Memory: ${Math.round(process.memoryUsage().rss / 1024 / 1024)}MB`);
    }, 200);

    async function finalize() {
        clearInterval(uiInterval);
        const duration = (performance.now() - startTime) / 1000;
        
        // Count unique IPs using Hamming Weight (SWAR)
        let uniqueCount = 0;
        for (let i = 0; i < bitsetView.length; i++) {
            let n = bitsetView[i];
            if (n === 0) continue;
            n = n - ((n >> 1) & 0x55555555);
            n = (n & 0x33333333) + ((n >> 2) & 0x33333333);
            uniqueCount += (((n + (n >> 4)) & 0x0F0F0F0F) * 0x01010101) >> 24;
        }

        console.log(`\n\n✅ SCAN COMPLETE`);
        console.log(`⏱️  Time:         ${duration.toFixed(1)}s`);
        console.log(`🏎️  Throughput:   ${(stats.size / 1024 / 1024 / duration).toFixed(0)} MB/s`);
        console.log(`🆔 Unique IPs:   ${uniqueCount.toLocaleString()}`);
        
        fs.closeSync(fd);
        if (uniqueCount>0) await saveResults();
		if (isTemp && fs.existsSync(inputFile)) {fs.unlinkSync(inputFile)};
        process.exit(0);
    }

async function saveResults() {
    console.log("📝 Writing to unique_ips.txt");
    const writeStream = fs.createWriteStream('unique_ips.txt');
    
    const buffer = Buffer.allocUnsafe(128 * 1024);
    let offset = 0;

    for (let i = 0; i < bitsetView.length; i++) {
        const word = bitsetView[i];
        if (word === 0) continue;

        for (let bit = 0; bit < 32; bit++) {
            if (word & (1 << bit)) {
                const ip = (i << 5) + bit;
                const s = `${(ip >>> 24)}.${(ip >> 16) & 255}.${(ip >> 8) & 255}.${ip & 255}\n`;
                const bytesWritten = buffer.write(s, offset, 'utf8');
                offset += bytesWritten;

                if (offset > 120 * 1024) {
                    if (!writeStream.write(buffer.subarray(0, offset))) {
                        await new Promise(r => writeStream.once('drain', r));
                    }
                    offset = 0;
                }
            }
        }
    }

    if (offset > 0) writeStream.write(buffer.subarray(0, offset));
    
    return new Promise(r => writeStream.end(() => {
        console.log("✅ Done.");
        r();
    }));
}




} else {
    /**
     * WORKER LOGIC
     */
    let bitset;
    parentPort.on('message', (msg) => {
        if (msg.type === 'start') {
            bitset = new Uint32Array(msg.bitsetSAB);
            parentPort.postMessage({ type: 'ready' });
            return;
        }

        const { sab, length } = msg;
        const buffer = new Uint8Array(sab);
        let start = -1;
        let lastMatchEnd = -1;

        for (let i = 0; i < length; i++) {
            const byte = buffer[i];
            const isIPChar = (byte >= 48 && byte <= 57) || byte === 46;

            if (isIPChar) {
                if (start === -1) start = i;
            } else if (start !== -1) {
                const len = i - start;
                if (len >= 7 && len <= 15) {
                    let dots = 0, res = 0, octet = 0, valid = true;
                    for (let j = start; j < i; j++) {
                        const b = buffer[j];
                        if (b === 46) {
                            dots++;
                            if (octet > 255) { valid = false; break; }
                            res = (res << 8) | octet;
                            octet = 0;
                        } else {
                            octet = octet * 10 + (b - 48);
                        }
                    }
                    if (valid && dots === 3 && octet <= 255) {
                        const ip = ((res << 8) | octet) >>> 0;
                        const wordIdx = ip >>> 5;
                        const bitIdx = ip & 31;
                        Atomics.or(bitset, wordIdx, 1 << bitIdx);
                        lastMatchEnd = i;
                    }
                }
                start = -1;
            }
        }
        parentPort.postMessage({ type: 'result', lastMatchEnd });
    });
}