streamflow/backend/routes/stream.js

const express = require('express');
const router = express.Router();
const { spawn, exec } = require('child_process');
const axios = require('axios');
const https = require('https');
const { SocksProxyAgent } = require('socks-proxy-agent');
const { HttpsProxyAgent } = require('https-proxy-agent');
const { authenticate } = require('../middleware/auth');
const { heavyLimiter, readLimiter } = require('../middleware/rateLimiter');
const { db } = require('../database/db');
const logger = require('../utils/logger');

// HTTPS agent to bypass SSL certificate verification for IPTV streams
const httpsAgent = new https.Agent({
  rejectUnauthorized: false
});

// Check if user has active VPN connection and return appropriate agent
const getVPNAgent = async (userId) => {
  return new Promise((resolve) => {
    db.get(
      'SELECT connected FROM vpn_settings WHERE user_id = ? AND connected = 1',
      [userId],
      (err, row) => {
        if (err || !row) {
          // No VPN, use standard HTTPS agent
          resolve(httpsAgent);
        } else {
          // VPN active - traffic will automatically route through VPN interface
          // Use standard agent, OS routing will handle VPN
          logger.info('VPN active for user, traffic will route through VPN');
          resolve(httpsAgent);
        }
      }
    );
  });
};

// Check hardware acceleration availability
const checkHardwareAcceleration = () => {
  const capabilities = {
    quicksync: false,
    nvenc: false,
    vaapi: false,
    videotoolbox: false
  };
  
  // Check for Intel Quick Sync (typically /dev/dri/renderD128)
  try {
    const fs = require('fs');
    if (fs.existsSync('/dev/dri/renderD128')) {
      capabilities.quicksync = true;
      capabilities.vaapi = true;
    }
  } catch (err) {
    logger.debug('Quick Sync not available');
  }
  
  // Check for NVIDIA
  try {
    const { execSync } = require('child_process');
    execSync('nvidia-smi', { stdio: 'ignore' });
    capabilities.nvenc = true;
  } catch (err) {
    logger.debug('NVENC not available');
  }
  
  return capabilities;
};

// Get hardware acceleration capabilities
router.get('/capabilities', authenticate, readLimiter, (req, res) => {
  const capabilities = checkHardwareAcceleration();
  res.json(capabilities);
});

// Get user's stream settings
const getStreamSettings = (userId, callback) => {
  db.get(
    'SELECT value FROM settings WHERE user_id = ? AND key = ?',
    [userId, 'stream_settings'],
    (err, result) => {
      if (err || !result) {
        // Default settings
        return callback({
          hwaccel: 'auto',
          hwaccel_device: '/dev/dri/renderD128',
          codec: 'h264',
          preset: 'veryfast',
          buffer_size: '2M',
          max_bitrate: '8M'
        });
      }
      
      try {
        callback(JSON.parse(result.value));
      } catch {
        callback({
          hwaccel: 'auto',
          hwaccel_device: '/dev/dri/renderD128',
          codec: 'h264',
          preset: 'veryfast',
          buffer_size: '2M',
          max_bitrate: '8M'
        });
      }
    }
  );
};

// Universal proxy for all streams with geo-blocking bypass
router.get('/proxy/:channelId', authenticate, heavyLimiter, async (req, res) => {
  const { channelId } = req.params;
  console.log(`[STREAM] Proxy request for channel ${channelId}`);
  
  try {
    const channel = await new Promise((resolve, reject) => {
      db.get('SELECT url, name, is_radio FROM channels WHERE id = ?', [channelId], (err, row) => {
        if (err) {
          console.error('[STREAM] Database error:', err);
          reject(err);
        } else if (!row) {
          console.error('[STREAM] Channel not found:', channelId);
          reject(new Error('Channel not found'));
        } else {
          console.log('[STREAM] Found channel:', row.name);
          resolve(row);
        }
      });
    });

    if (!channel.url) {
      return res.status(400).json({ error: 'Channel has no URL' });
    }

    logger.info(`Proxying ${channel.is_radio ? 'radio' : 'video'} stream: ${channel.name} - ${channel.url}`);

    // Extract origin from URL for proper spoofing
    const urlObj = new URL(channel.url);
    const origin = `${urlObj.protocol}//${urlObj.hostname}`;
    const referer = origin;

    // Get VPN-aware agent for this user
    const agent = await getVPNAgent(req.user.userId);

    const requestConfig = {
      method: 'GET',
      url: channel.url,
      responseType: channel.url.includes('.m3u8') ? 'text' : 'stream',
      timeout: 30000,
      validateStatus: (status) => status < 500,
      httpsAgent: agent,
      headers: {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
        'Accept': '*/*',
        'Accept-Language': 'en-US,en;q=0.9',
        'Accept-Encoding': 'identity',
        'Origin': origin,
        'Referer': referer,
        'Connection': 'keep-alive',
        'Sec-Fetch-Dest': 'empty',
        'Sec-Fetch-Mode': 'cors',
        'Sec-Fetch-Site': 'cross-site'
      }
    };

    // Proxy with origin spoofing to bypass geo-blocking
    const response = await axios(requestConfig);

    if (response.status >= 400) {
      logger.error(`Stream returned status ${response.status}`);
      return res.status(response.status).json({ error: 'Stream unavailable' });
    }

    // Handle HLS manifests - rewrite URLs to go through our proxy
    if (channel.url.includes('.m3u8')) {
      const m3u8Content = response.data;
      const baseUrl = channel.url.substring(0, channel.url.lastIndexOf('/') + 1);
      
      // Rewrite relative URLs in playlist to absolute URLs through our proxy
      const rewrittenContent = m3u8Content.split('\n').map(line => {
        if (line.startsWith('#') || line.trim() === '') {
          return line;
        }
        // Convert relative to absolute URL
        let absoluteUrl = line;
        if (!line.startsWith('http')) {
          absoluteUrl = baseUrl + line;
        }
        // Proxy the URL through our server
        const proxyUrl = `/api/stream/hls-segment?url=${encodeURIComponent(absoluteUrl)}&token=${req.query.token}`;
        return proxyUrl;
      }).join('\n');

      res.setHeader('Content-Type', 'application/vnd.apple.mpegurl');
      res.setHeader('Access-Control-Allow-Origin', '*');
      res.setHeader('Access-Control-Allow-Headers', '*');
      res.setHeader('Cache-Control', 'no-cache');
      return res.send(rewrittenContent);
    }

    // For binary streams, pipe directly
    res.setHeader('Content-Type', response.headers['content-type'] || 'application/octet-stream');
    res.setHeader('Cache-Control', 'no-cache');
    res.setHeader('Access-Control-Allow-Origin', '*');
    res.setHeader('Access-Control-Allow-Headers', '*');
    res.setHeader('Access-Control-Allow-Methods', 'GET, HEAD, OPTIONS');
    
    if (response.headers['content-length']) {
      res.setHeader('Content-Length', response.headers['content-length']);
    }
    if (response.headers['content-range']) {
      res.setHeader('Content-Range', response.headers['content-range']);
      res.setHeader('Accept-Ranges', 'bytes');
    }

    response.data.pipe(res);

    response.data.on('error', (error) => {
      logger.error('Stream error:', error.message);
      if (!res.headersSent) {
        res.status(500).json({ error: 'Stream failed' });
      }
    });

    req.on('close', () => {
      logger.info('Client disconnected from stream');
      if (response.data && !response.data.destroyed) {
        response.data.destroy();
      }
    });

  } catch (error) {
    console.error('[STREAM] Proxy error:', error);
    logger.error('Proxy error:', error.message);
    if (!res.headersSent) {
      res.status(500).json({ error: 'Failed to proxy stream', details: error.message });
    }
  }
});

// Proxy HLS segments (playlists and .ts chunks)
router.get('/hls-segment', authenticate, heavyLimiter, async (req, res) => {
  const { url } = req.query;
  
  if (!url) {
    return res.status(400).json({ error: 'URL parameter required' });
  }

  try {
    const urlObj = new URL(url);
    const origin = `${urlObj.protocol}//${urlObj.hostname}`;

    logger.info(`Proxying HLS segment: ${url}`);

    const response = await axios({
      method: 'GET',
      url: url,
      responseType: url.includes('.m3u8') ? 'text' : 'stream',
      timeout: 15000,
      httpsAgent: httpsAgent,
      validateStatus: (status) => status < 500,
      headers: {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
        'Accept': '*/*',
        'Origin': origin,
        'Referer': origin,
        'Connection': 'keep-alive'
      }
    });

    if (response.status >= 400) {
      return res.status(response.status).json({ error: 'Segment unavailable' });
    }

    // Handle nested m3u8 playlists
    if (url.includes('.m3u8')) {
      const m3u8Content = response.data;
      const baseUrl = url.substring(0, url.lastIndexOf('/') + 1);
      
      const rewrittenContent = m3u8Content.split('\n').map(line => {
        if (line.startsWith('#') || line.trim() === '') {
          return line;
        }
        let absoluteUrl = line;
        if (!line.startsWith('http')) {
          absoluteUrl = baseUrl + line;
        }
        return `/api/stream/hls-segment?url=${encodeURIComponent(absoluteUrl)}&token=${req.query.token}`;
      }).join('\n');

      res.setHeader('Content-Type', 'application/vnd.apple.mpegurl');
      res.setHeader('Access-Control-Allow-Origin', '*');
      res.setHeader('Cache-Control', 'no-cache');
      return res.send(rewrittenContent);
    }

    // Stream binary segments
    res.setHeader('Content-Type', response.headers['content-type'] || 'video/mp2t');
    res.setHeader('Access-Control-Allow-Origin', '*');
    res.setHeader('Cache-Control', 'public, max-age=300');
    
    if (response.headers['content-length']) {
      res.setHeader('Content-Length', response.headers['content-length']);
    }

    response.data.pipe(res);

    response.data.on('error', (error) => {
      logger.error('HLS segment error:', error.message);
      if (!res.headersSent) {
        res.status(500).end();
      }
    });

  } catch (error) {
    logger.error('HLS segment proxy error:', error.message);
    if (!res.headersSent) {
      res.status(500).json({ error: 'Failed to proxy segment' });
    }
  }
});

// Stream proxy with hardware acceleration (for transcoding if needed)
router.get('/proxy-ffmpeg/:channelId', authenticate, heavyLimiter, (req, res) => {
  const { channelId } = req.params;
  
  db.get(
    'SELECT url FROM channels WHERE id = ?',
    [channelId],
    (err, channel) => {
      if (err || !channel) {
        return res.status(404).json({ error: 'Channel not found' });
      }
      
      getStreamSettings(req.user.userId, (settings) => {
        const capabilities = checkHardwareAcceleration();
        
        // Build FFmpeg command with hardware acceleration
        const ffmpegArgs = [
          '-re',
          '-i', channel.url,
          '-c:v', 'copy',
          '-c:a', 'copy',
          '-f', 'mpegts',
          'pipe:1'
        ];
        
        // Add hardware acceleration if enabled and available
        if (settings.hwaccel !== 'none') {
          if (settings.hwaccel === 'quicksync' && capabilities.quicksync) {
            ffmpegArgs.unshift(
              '-hwaccel', 'qsv',
              '-hwaccel_device', settings.hwaccel_device || '/dev/dri/renderD128',
              '-hwaccel_output_format', 'qsv'
            );
          } else if (settings.hwaccel === 'vaapi' && capabilities.vaapi) {
            ffmpegArgs.unshift(
              '-hwaccel', 'vaapi',
              '-hwaccel_device', settings.hwaccel_device || '/dev/dri/renderD128',
              '-hwaccel_output_format', 'vaapi'
            );
          } else if (settings.hwaccel === 'nvenc' && capabilities.nvenc) {
            ffmpegArgs.unshift(
              '-hwaccel', 'cuda',
              '-hwaccel_output_format', 'cuda'
            );
          } else if (settings.hwaccel === 'auto') {
            // Auto-detect best available
            if (capabilities.quicksync) {
              ffmpegArgs.unshift(
                '-hwaccel', 'qsv',
                '-hwaccel_device', '/dev/dri/renderD128',
                '-hwaccel_output_format', 'qsv'
              );
            } else if (capabilities.nvenc) {
              ffmpegArgs.unshift(
                '-hwaccel', 'cuda',
                '-hwaccel_output_format', 'cuda'
              );
            }
          }
        }
        
        logger.info(`Streaming channel ${channelId} with args:`, ffmpegArgs);
        
        const ffmpeg = spawn('ffmpeg', ffmpegArgs);
        
        res.setHeader('Content-Type', 'video/mp2t');
        res.setHeader('Cache-Control', 'no-cache');
        
        ffmpeg.stdout.pipe(res);
        
        ffmpeg.stderr.on('data', (data) => {
          logger.debug(`FFmpeg: ${data}`);
        });
        
        ffmpeg.on('error', (error) => {
          logger.error('FFmpeg error:', error);
          if (!res.headersSent) {
            res.status(500).json({ error: 'Stream failed' });
          }
        });
        
        ffmpeg.on('close', (code) => {
          logger.info(`FFmpeg closed with code ${code}`);
        });
        
        req.on('close', () => {
          logger.info('Client disconnected, stopping stream');
          ffmpeg.kill('SIGTERM');
        });
      });
    }
  );
});

module.exports = router;