const WebSocket = require('ws'); const fs = require('fs'); const readline =
require('readline'); const path = require('path');
/ Configuration
const PORT = process.env.PORT || 8080; const ARCHIVE_FILE = path.join(__dirname
, 'mesh_archive.jsonl');
/ Materialized view of the VFS in RAM
/ Map<string, { data: string, timestamp: number }>
const vfsState = new Map();
/ Open an append-only stream for efficient, ordered, concurrent-safe disk
writes
const archiveStream = fs.createWriteStream(ARCHIVE_FILE, { flags: 'a' });
async function rebuildStateFromLog() {
    if (!fs.existsSync(ARCHIVE_FILE)) {
        console.log(`[dVFS] No existing archive found at ${ARCHIVE_FILE}.
Starting fresh.`);
        return;
    }
    console.log(`[dVFS] Rebuilding state from ${ARCHIVE_FILE}...`);
    
    const fileStream = fs.createReadStream(ARCHIVE_FILE);
    const rl = readline.createInterface({
        input: fileStream,
        crlfDelay: Infinity
    });
    let eventCount = 0;
    for await (const line of rl) {
        if (!line.trim()) continue;
        
        try {
            const event = JSON.parse(line);
            applyEventToRAM(event);
            eventCount++;
        } catch (err) {
            console.error(`[dVFS] Corrupted log entry skipped: ${line}`,
err.message);
        }
    }
    console.log(`[dVFS] State rebuilt successfully. Replayed ${eventCount}
events. Tracking ${vfsState.size} active files.`);
}
function applyEventToRAM(event) {
    const { type, path: filePath, data, timestamp } = event;
    const current = vfsState.get(filePath);
    // 1. Stale Update Check (Out-of-order packet protection)
    if (current && current.timestamp > timestamp) {
        return false; 
    }
    // 2. Idempotency Check (Content hasn't changed)
    if (type === 'VFS_WRITE') {
        if (current && current.data === data) {
            return false;
        }
        vfsState.set(filePath, { data, timestamp });
        return true;
    } 
    
    if (type === 'VFS_DELETE') {
        if (!current) {
            return false; // Already deleted / doesn't exist
        }
        vfsState.delete(filePath);
        return true;
    }
    return false;
}
function startServer() {
    const wss = new WebSocket.Server({ port: PORT });
    wss.on('connection', (ws, req) => {
        const clientIp = req.socket.remoteAddress;
        console.log(`[dVFS] Client connected from ${clientIp}`);
        // Immediately push the current materialized state to the new client
        vfsState.forEach((fileInfo, filePath) => {
            const syncEvent = {
                type: 'VFS_WRITE',
                path: filePath,
                data: fileInfo.data,
                timestamp: fileInfo.timestamp
            };
            ws.send(JSON.stringify(syncEvent));
        });
        ws.on('message', (messageAsString) => {
            let event;
            try {
                event = JSON.parse(messageAsString);
            } catch (e) {
                console.warn(`[dVFS] Received malformed JSON from
${clientIp}`);
                return;
            }
            // Schema Validation
            if (!event.type || typeof event.path !== 'string' || typeof
event.timestamp !== 'number') {
                console.warn(`[dVFS] Received invalid payload schema from
${clientIp}:`, event);
                return;
            }
            if (event.type === 'VFS_WRITE' && typeof event.data !== 'string')
{
                console.warn(`[dVFS] VFS_WRITE missing string data from
${clientIp}`);
                return;
            }
            // Apply to RAM with idempotency & LWW checks
            const didMutate = applyEventToRAM(event);
            if (didMutate) {
                // Persist to Append-Only Log
                // We stringify the exact event to maintain consistent schema
on disk
                archiveStream.write(JSON.stringify(event) + '
');
                // Broadcast to all *other* connected clients
                wss.clients.forEach((client) => {
                    if (client !== ws && client.readyState === WebSocket.OPEN)
{
                        client.send(messageAsString); // relay original string
buffer for perf
                    }
                });
                
                console.log(`[dVFS] Processed & Broadcasted ${event.type} for
${event.path} (ts: ${event.timestamp})`);
            }
        });
        ws.on('close', () => {
            console.log(`[dVFS] Client disconnected: ${clientIp}`);
        });
        
        ws.on('error', (err) => {
            console.error(`[dVFS] WebSocket error for ${clientIp}:`,
err.message);
        });
    });
    console.log(`[dVFS] Relay Node running on ws://localhost:${PORT}`);
}
/ Boot Sequence
rebuildStateFromLog()
    .then(startServer)
    .catch((err) => {
        console.error('[dVFS] FATAL: Failed to initialize relay node', err);
        process.exit(1);
    });
/ Graceful shutdown handling
process.on('SIGINT', () => {
    console.log('
[dVFS] Shutting down relay node...');
    archiveStream.end();
    process.exit(0);
});