server.js

require('dotenv').config();

const express = require('express');
const WebSocket = require('ws');
const { spawn } = require('child_process');
const http = require('http');

// ================= CONFIG =================
const PORT = process.env.PORT || 7860;

const STREAM_KEY = process.env.STREAM_KEY;

const AUTH_TOKEN = process.env.AUTH_TOKEN;

let rawWeb =
    process.env.WEB ||
    'webbb-production-b681.up.railway.app';

const ALLOWED_ORIGIN_HOST = rawWeb
    .replace(/^https?:\/\//, '')
    .replace(/\/$/, '');

const RTMP_URL =
    `rtmp://a.rtmp.youtube.com/live2/${STREAM_KEY}`;

if (!STREAM_KEY || !AUTH_TOKEN) {

    console.error(
        '❌ Missing STREAM_KEY or AUTH_TOKEN'
    );

    process.exit(1);
}

// ================= QUALITY LEVELS =================
// محسنة للاستضافات المجانية

const QUALITY_LEVELS = [

    {
        name: '1080p',
        width: 1920,
        height: 1080,
        bitrate: 3500,
        audio: 128
    },

    {
        name: '720p',
        width: 1280,
        height: 720,
        bitrate: 2200,
        audio: 128
    },

    {
        name: '480p',
        width: 854,
        height: 480,
        bitrate: 1000,
        audio: 96
    }
];

// ================= STATE =================
const state = {

    ffmpeg: null,

    activeClient: null,

    isStreaming: false,

    lastDataTime: 0,

    startTime: 0,

    restarting: false,

    cleaning: false,

    // البداية 720p أفضل شيء
    qualityIndex: 1,

    // مراقبة الضغط
    bufferWarnings: 0
};

function getQuality() {

    return QUALITY_LEVELS[
        state.qualityIndex
    ];
}

const app = express();

const server = http.createServer(app);

// ================= HEALTH =================
app.get('/', (_, res) => {

    res.send(
        '🚀 Titan Engine Online'
    );
});

app.get('/health', (_, res) => {

    const q = getQuality();

    res.json({

        streaming: state.isStreaming,

        quality: q.name,

        uptime: state.startTime
            ? Math.floor(
                (
                    Date.now() -
                    state.startTime
                ) / 1000
            )
            : 0,

        ffmpeg: !!state.ffmpeg
    });
});

// ================= CLEANUP =================
async function deepCleanup() {

    if (state.cleaning) return;

    state.cleaning = true;

    console.log(
        '🧹 Cleaning session...'
    );

    try {

        state.isStreaming = false;

        if (state.activeClient) {

            try {

                state.activeClient.terminate();

            } catch {}
        }

        state.activeClient = null;

        if (state.ffmpeg) {

            try {

                if (state.ffmpeg.stdin) {

                    state.ffmpeg.stdin.destroy();
                }

            } catch {}

            try {

                if (
                    process.platform === 'win32'
                ) {

                    state.ffmpeg.kill(
                        'SIGKILL'
                    );

                } else {

                    process.kill(
                        -state.ffmpeg.pid,
                        'SIGKILL'
                    );
                }

            } catch {

                try {

                    state.ffmpeg.kill(
                        'SIGKILL'
                    );

                } catch {}
            }

            state.ffmpeg = null;
        }

        console.log(
            '✅ Session cleaned'
        );

    } finally {

        state.cleaning = false;
    }
}

// ================= RESTART =================
function safeRestartFFmpeg() {

    if (
        state.restarting ||
        state.cleaning
    ) return;

    state.restarting = true;

    console.log(
        '🔄 Restarting FFmpeg...'
    );

    setTimeout(() => {

        if (state.isStreaming) {

            startFFmpeg();
        }

        state.restarting = false;

    }, 2000);
}

// ================= ADAPTIVE QUALITY =================
async function checkAdaptiveQuality() {

    if (
        !state.ffmpeg ||
        !state.ffmpeg.stdin
    ) return;

    const buffer =
        state.ffmpeg.stdin.writableLength || 0;

    // النت ضعيف
    if (buffer > 2 * 1024 * 1024) {

        state.bufferWarnings++;

    } else {

        state.bufferWarnings = 0;
    }

    // downgrade
    if (state.bufferWarnings >= 3) {

        if (
            state.qualityIndex <
            QUALITY_LEVELS.length - 1
        ) {

            state.qualityIndex++;

            console.log(
                `📉 Weak internet → ${getQuality().name}`
            );

            await restartEncoder();
        }

        state.bufferWarnings = 0;
    }

    // upgrade
    if (
        buffer < 300 * 1024 &&
        state.qualityIndex > 0
    ) {

        state.qualityIndex--;

        console.log(
            `📈 Strong internet → ${getQuality().name}`
        );

        await restartEncoder();
    }
}

async function restartEncoder() {

    if (state.ffmpeg) {

        try {

            state.ffmpeg.stdin.destroy();

        } catch {}

        try {

            if (
                process.platform === 'win32'
            ) {

                state.ffmpeg.kill(
                    'SIGKILL'
                );

            } else {

                process.kill(
                    -state.ffmpeg.pid,
                    'SIGKILL'
                );
            }

        } catch {}
    }

    safeRestartFFmpeg();
}

// ================= FFMPEG =================
function startFFmpeg() {

    const q = getQuality();

    console.log(
        `🎬 Starting ${q.name}`
    );

    const args = [

        '-loglevel', 'error',

        // latency optimization
        '-fflags', 'nobuffer',

        '-flags', 'low_delay',

        '-strict', 'experimental',

        '-max_muxing_queue_size', '1024',

        '-i', 'pipe:0',

        // ================= VIDEO =================

        '-c:v', 'libx264',

        // أقل استهلاك CPU
        '-preset', 'ultrafast',

        '-tune', 'zerolatency',

        // جودة أفضل من baseline
        '-profile:v', 'main',

        '-threads', '1',

        '-pix_fmt', 'yuv420p',

        '-vf',
        `scale=${q.width}:${q.height}`,

        '-r', '30',

        '-g', '60',

        '-keyint_min', '60',

        '-sc_threshold', '0',

        '-b:v',
        `${q.bitrate}k`,

        '-maxrate',
        `${q.bitrate}k`,

        '-bufsize',
        `${q.bitrate * 2}k`,

        // ================= AUDIO =================

        '-c:a', 'aac',

        '-b:a',
        `${q.audio}k`,

        '-ar', '44100',

        '-ac', '2',

        // ================= OUTPUT =================

        '-f', 'flv',

        RTMP_URL
    ];

    state.ffmpeg = spawn(
        'ffmpeg',
        args,
        {
            detached:
                process.platform !== 'win32'
        }
    );

    state.ffmpeg.stdin.on(
        'error',
        () => {}
    );

    state.ffmpeg.stderr.on(
        'data',
        () => {}
    );

    state.ffmpeg.on(
        'close',
        code => {

            console.log(
                `📡 FFmpeg closed (${code})`
            );

            if (
                state.isStreaming &&
                !state.restarting &&
                !state.cleaning
            ) {

                safeRestartFFmpeg();
            }
        }
    );
}

// ================= WEBSOCKET =================
const wss = new WebSocket.Server({

    noServer: true,

    maxPayload: 1024 * 1024
});

server.on(
    'upgrade',
    (req, socket, head) => {

        const url = new URL(
            req.url,
            `http://${req.headers.host}`
        );

        const token =
            url.searchParams.get('token');

        if (token !== AUTH_TOKEN) {

            console.log(
                '🚫 Invalid token'
            );

            socket.write(
                'HTTP/1.1 401 Unauthorized\r\n\r\n'
            );

            socket.destroy();

            return;
        }

        const origin =
            req.headers.origin || '';

        let hostname = '';

        try {

            hostname =
                new URL(origin).hostname;

        } catch {}

        const isLocal =
            hostname === 'localhost' ||
            hostname === '127.0.0.1';

        if (
            origin &&
            hostname !==
            ALLOWED_ORIGIN_HOST &&
            !isLocal
        ) {

            console.log(
                `🚫 Blocked origin: ${hostname}`
            );

            socket.write(
                'HTTP/1.1 403 Forbidden\r\n\r\n'
            );

            socket.destroy();

            return;
        }

        wss.handleUpgrade(
            req,
            socket,
            head,
            ws => wss.emit(
                'connection',
                ws
            )
        );
    }
);

// ================= CONNECTION =================
wss.on(
    'connection',
    async ws => {

        console.log(
            '✅ Teacher connected'
        );

        if (state.isStreaming) {

            console.log(
                '🧹 Replacing old session'
            );

            await deepCleanup();
        }

        state.activeClient = ws;

        state.isStreaming = true;

        state.startTime = Date.now();

        state.lastDataTime = Date.now();

        ws.isAlive = true;

        startFFmpeg();

        ws.on(
            'pong',
            () => {

                ws.isAlive = true;
            }
        );

        ws.on(
            'message',
            data => {

                if (
                    !state.ffmpeg ||
                    !state.ffmpeg.stdin ||
                    !state.ffmpeg.stdin.writable
                ) {
                    return;
                }

                state.lastDataTime =
                    Date.now();

                // حماية الرام
                if (
                    state.ffmpeg.stdin
                        .writableLength >
                    3 * 1024 * 1024
                ) {

                    console.log(
                        '⚠️ Buffer pressure'
                    );

                    return;
                }

                const ok =
                    state.ffmpeg.stdin
                        .write(data);

                // backpressure
                if (
                    !ok &&
                    ws._socket
                ) {

                    ws._socket.pause();

                    state.ffmpeg.stdin.once(
                        'drain',
                        () => {

                            if (
                                ws.readyState ===
                                WebSocket.OPEN &&
                                ws._socket
                            ) {

                                ws._socket.resume();
                            }
                        }
                    );
                }
            }
        );

        ws.on(
            'close',
            async () => {

                console.log(
                    '🔌 Teacher disconnected'
                );

                await deepCleanup();
            }
        );

        ws.on(
            'error',
            async err => {

                console.log(
                    '❌ WS Error:',
                    err.message
                );

                await deepCleanup();
            }
        );
    }
);

// ================= WATCHDOGS =================

// idle timeout
setInterval(
    async () => {

        if (
            state.isStreaming &&
            Date.now() -
            state.lastDataTime >
            20000
        ) {

            console.log(
                '🕒 Idle timeout'
            );

            await deepCleanup();
        }

    },
    10000
);

// ping pong
setInterval(
    () => {

        wss.clients.forEach(
            ws => {

                if (!ws.isAlive) {

                    try {

                        ws.terminate();

                    } catch {}

                    return;
                }

                ws.isAlive = false;

                try {

                    ws.ping();

                } catch {}
            }
        );

    },
    15000
);

// adaptive quality
setInterval(
    async () => {

        if (!state.isStreaming) return;

        await checkAdaptiveQuality();

    },
    12000
);

// stats
setInterval(
    () => {

        if (!state.isStreaming) return;

        const q = getQuality();

        const uptime =
            Math.floor(
                (
                    Date.now() -
                    state.startTime
                ) / 1000
            );

        const buffer =
            state.ffmpeg?.stdin
                ?.writableLength || 0;

        console.log(

            `📊 LIVE | ` +
            `${q.name} | ` +
            `${q.bitrate}kbps | ` +
            `${uptime}s | ` +
            `${(buffer / 1024).toFixed(0)}KB`
        );

    },
    30000
);

// ================= ERROR HANDLING =================
process.on(
    'uncaughtException',
    err => {

        console.error(
            '🔥 Error:',
            err.message
        );
    }
);

process.on(
    'SIGINT',
    async () => {

        await deepCleanup();

        process.exit(0);
    }
);

process.on(
    'SIGTERM',
    async () => {

        await deepCleanup();

        process.exit(0);
    }
);

// ================= START =================
server.listen(
    PORT,
    () => {

        console.log(
            `🚀 Titan Engine running on ${PORT}`
        );

        console.log(
            `📡 YouTube Live Ready`
        );

        console.log(
            `🎥 Default Quality: ${getQuality().name}`
        );
    }
);