universal_is/monitor/index.ts

import { serve } from 'bun';
import { Database } from 'bun:sqlite';
import { EventEmitter } from 'events';

const db = new Database('./data/servers.db');

// Initialize the database table if it doesn't exist
db.run(`
    CREATE TABLE IF NOT EXISTS servers (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        hostname TEXT NOT NULL,
        port INTEGER NOT NULL,
        ping_rate INTEGER NOT NULL,
        status TEXT DEFAULT 'unknown',
        monitoring INTEGER DEFAULT 1
    )
`);

// In-memory state to store server details
const serverStates: {
    [id: number]: {
        id: number;
        name: string;
        hostname: string;
        port: number;
        ping_rate: number;
        status: 'up' | 'down' | 'unknown';
        monitoring: number;
    }
} = {};

// Event emitter to handle server updates
const serverMonitor = new EventEmitter();
const clients: Set<ReadableStreamDefaultController> = new Set()

class ServerState {
    id: number | undefined;
    name: string | undefined;
    hostname: string | undefined;
    port: number | undefined;
    ping_rate: number | undefined;
    status: 'up' | 'down' | 'unknown' | undefined;
    monitoring: number | undefined;
}

// Function to initialize server states from the database
async function initializeServerStates() {
    const servers = db.query('SELECT * FROM servers').as(ServerState).all();
    servers.forEach((server: ServerState) => {
        if (server.id && server.name && server.hostname && server.port && server.ping_rate && server.status) {
            serverStates[server.id] = {
                id: server.id,
                name: server.name,
                hostname: server.hostname,
                port: server.port,
                ping_rate: server.ping_rate,
                status: server.status,
                monitoring: server.monitoring || 0
            };
        }
    });
}

// Start monitoring for a specific server by its ID
async function monitorServer(serverId: number) {
    if (!serverStates[serverId]) return

    while (true) {
        if (serverStates[serverId]) {
            await checkServer(serverId);
            await new Promise(resolve => setTimeout(resolve, serverStates[serverId].ping_rate * 1000));
        } else {
            console.log("Stopped monitoring ", serverId)
            return false
        }
    }
}

// Function to check the server availability
async function checkServer(serverId: number) {
    const server = serverStates[serverId];
    if (!server) return;
    if (server.monitoring === 0) return

    try {
        const checkSocket = await Bun.connect({
            hostname: server.hostname,
            port: server.port,
            socket: {
                data(socket, data) { },
                open(socket) {
                    console.log(`[${new Date().toISOString()}] ${server.name} (${server.hostname}:${server.port}) is up.`);

                    if (serverStates[serverId].status !== 'up') {
                        db.run('UPDATE servers SET status = ? WHERE id = ?', ['up', serverId]);
                        serverStates[serverId].status = 'up';
                        notifyClients()
                    }
                    socket.end();
                },
                connectError(socket, error) {
                    console.error(`[${new Date().toISOString()}] ${server.name} (${server.hostname}:${server.port}) is down!`);

                    if (serverStates[serverId].status !== 'down') {
                        db.run('UPDATE servers SET status = ? WHERE id = ?', ['down', serverId]);
                        serverStates[serverId].status = 'down';
                        notifyClients()
                    }
                },
                error(socket, error) {
                    console.error(`[${new Date().toISOString()}] Error connecting to ${server.name} (${server.hostname}:${server.port}):`, error);

                    if (serverStates[serverId].status !== 'down') {
                        db.run('UPDATE servers SET status = ? WHERE id = ?', ['down', serverId]);
                        serverStates[serverId].status = 'down';
                        notifyClients()
                    }
                },
                timeout(socket) {
                    console.error(`[${new Date().toISOString()}] Connection to ${server.name} (${server.hostname}:${server.port}) timed out!`);

                    if (serverStates[serverId].status !== 'down') {
                        db.run('UPDATE servers SET status = ? WHERE id = ?', ['down', serverId]);
                        serverStates[serverId].status = 'down';
                        notifyClients()
                    }
                }
            }
        })
    } catch (error) {
        console.error(`[${new Date().toISOString()}] Unexpected error checking ${server.name} (${server.hostname}:${server.port}):`, error);

        if (serverStates[serverId].status !== 'down') {
            db.run('UPDATE servers SET status = ? WHERE id = ?', ['down', serverId]);
            serverStates[serverId].status = 'down';
        }
    }
}

// Notify all connected clients of server state changes
function notifyClients() {
    const data = JSON.stringify(Object.values(serverStates));
    clients.forEach(client => client.enqueue(`data: ${data}\n\n`));
}

// Start monitoring all servers from the initial state
async function startMonitoring() {
    Object.keys(serverStates).forEach(serverId => {
        monitorServer(parseInt(serverId, 10));
    })
    console.log('Monitoring started for existing servers.');
}

// Initialize states and start monitoring
initializeServerStates().then(startMonitoring);

// Event listener for adding new servers
serverMonitor.on('newServer', (serverId) => {
    if (!serverStates[serverId]) return;
    monitorServer(serverId);
});

// API Server to manage servers
const server = serve({
    port: process.env.MONITOR_PORT || 1234,
    fetch: async (req) => {
        const url = new URL(req.url);
        const pathname = url.pathname;
        const method = req.method;

        if (pathname === '/watch' && method === 'GET') {
            const headers = {
                'Content-Type': 'text/event-stream',
                'Cache-Control': 'no-cache',
                'Connection': 'keep-alive',
                'Access-Control-Allow-Origin': '*'
            };
            const client = new Response(
                new ReadableStream({
                    start(controller) {
                        clients.add(controller);
                        controller.enqueue(`data: ${JSON.stringify(Object.values(serverStates))}\n\n`)
                    },
                    cancel(controller) {
                        clients.delete(controller);
                    }
                }),
                { headers }
            );
            return client;
        }

        if (pathname === '/servers' && method === 'GET') {
            return new Response(JSON.stringify(Object.values(serverStates)), {
                headers: { 'Content-Type': 'application/json', 'Access-Control-Allow-Origin': '*' },
                status: 200
            });
        }

        if (pathname === '/server' && method === 'POST') {
            const data = await req.json();
            const { name, hostname, port, ping_rate } = data;

            if (!name || !hostname || !port || !ping_rate) {
                return new Response('Missing fields', { status: 400 });
            }

            db.run('INSERT INTO servers (name, hostname, port, ping_rate) VALUES (?, ?, ?, ?)', name, hostname, port, ping_rate);
            const newServer = db.query('SELECT * FROM servers WHERE id = last_insert_rowid()').as(ServerState).get();

            if (newServer && newServer.id && newServer.name && newServer.hostname && newServer.port && newServer.ping_rate && newServer.status && newServer.monitoring) {
                serverStates[newServer.id] = {
                    id: newServer.id,
                    name: newServer.name,
                    hostname: newServer.hostname,
                    port: newServer.port,
                    ping_rate: newServer.ping_rate,
                    status: newServer.status || 'unknown',
                    monitoring: newServer.monitoring
                };
            }

            if (newServer && newServer.id) {
                serverMonitor.emit('newServer', newServer.id);
                return new Response('Server added', { status: 201, headers: { 'Access-Control-Allow-Origin': '*' } });
            }
        }

        if (pathname.startsWith('/server/') && method === 'PUT') {
            const { searchParams } = new URL(req.url)
            const id = pathname.split('/').pop()

            const name = searchParams.get('name')
            const hostname = searchParams.get('hostname')
            const port = searchParams.get('port')
            const ping_rate = searchParams.get('ping_rate')
            const status = searchParams.get('status')
            const monitoring = searchParams.get('monitoring')

            if (!id) {
                return new Response('Invalid ID', { status: 400, headers: { 'Access-Control-Allow-Origin': '*' } });
            }

            // Fetch the current state of the server
            const currentServer = serverStates[parseInt(id)];
            if (!currentServer) {
                return new Response('Server not found', { status: 404, headers: { 'Access-Control-Allow-Origin': '*' } });
            }

            // Prepare the updated server data
            const updatedServer: any = {
                name: name || currentServer.name,
                hostname: hostname || currentServer.hostname,
                port: port || currentServer.port,
                ping_rate: ping_rate || currentServer.ping_rate,
                status: status || currentServer.status,
                monitoring: monitoring || currentServer.monitoring
            };

            // Update the database
            db.run(
                `UPDATE servers SET name = ?, hostname = ?, port = ?, ping_rate = ?, status = ?, monitoring = ? WHERE id = ?`,
                updatedServer.name, updatedServer.hostname, updatedServer.port, updatedServer.ping_rate, updatedServer.status, updatedServer.monitoring, id
            );

            // Update the in-memory state
            serverStates[parseInt(id)] = {
                ...serverStates[parseInt(id)], // Retain existing fields
                ...updatedServer // Apply updates
            };

            return new Response('Server updated', { status: 200, headers: { 'Access-Control-Allow-Origin': '*' } });
        }

        if (pathname.startsWith('/server/') && method === 'DELETE') {
            const id = pathname.split('/').pop();
            if (id) {
                db.run('DELETE FROM servers WHERE id = ?', [parseInt(id)])
                delete serverStates[parseInt(id)]

                return new Response('Server deleted', { status: 200, headers: { 'Access-Control-Allow-Origin': '*' } })
            }
        }

        return new Response('Not Found', { status: 404, headers: { 'Access-Control-Allow-Origin': '*' } });
    }
});

console.log(`API server and monitoring running on http://localhost:${server.port}`);