Update server.js

server.js

@@ -2,13 +2,21 @@ const express = require('express');
 const fetch = require('node-fetch');
 const cors = require('cors');
 const rateLimit = require('express-rate-limit');
+const helmet = require('helmet');
 require('dotenv').config();
 
 const app = express();
 
+// --- SECURITY HEADERS ---
+app.use(helmet({
+    contentSecurityPolicy: false,
+    crossOriginEmbedderPolicy: false
+}));
+
 app.set('trust proxy', 1);
 app.use(express.json({ limit: '1mb' }));
 
+// --- CORS PROTECTION ---
 const allowedOrigins = process.env.ALLOWED_ORIGINS 
     ? process.env.ALLOWED_ORIGINS.split(',').map(o => o.trim())
     : [];
@@ -19,9 +27,11 @@ app.use(cors({
         if (allowedOrigins.includes(origin)) {
             callback(null, true);
         } else {
+            console.log(`[Security] Blocked origin: ${origin}`);
             callback(new Error('Not allowed by CORS'));
         }
-    }
+    },
+    credentials: true
 }));
 
 app.use((err, req, res, next) => {
@@ -31,165 +41,385 @@ app.use((err, req, res, next) => {
     next(err);
 });
 
+// --- RATE LIMITING (15 minutes) ---
 const limiter = rateLimit({
     windowMs: 15 * 60 * 1000,
     max: 100,
-    message: { error: "Too many requests" }
+    message: { error: "Too many requests" },
+    keyGenerator: (req) => {
+        const ip = req.ip || req.connection.remoteAddress || 'unknown';
+        return ip.replace(/:\d+[^:]*$/, '');
+    }
 });
 app.use(limiter);
 
+// --- REQUEST LOGGING ---
 app.use((req, res, next) => {
     const ip = (req.ip || 'unknown').replace(/:\d+[^:]*$/, '');
-    console.log(`[${new Date().toISOString()}] ${ip} -> ${req.method} ${req.path} from ${req.headers.origin || 'unknown'}`);
+    const origin = (req.headers.origin || 'unknown').substring(0, 100);
+    console.log(`[${new Date().toISOString()}] ${ip} -> ${req.method} ${req.path} from ${origin}`);
+    next();
+});
+
+// --- DAILY USAGE CAPS ---
+const dailyUsage = new Map();
+let lastResetDate = new Date().toDateString();
+
+function checkDailyReset() {
+    const today = new Date().toDateString();
+    if (today !== lastResetDate) {
+        dailyUsage.clear();
+        lastResetDate = today;
+        console.log('[System] Daily usage counters reset');
+    }
+}
+
+setInterval(checkDailyReset, 60 * 60 * 1000);
+
+app.use((req, res, next) => {
+    if (req.method === 'POST' && req.path.includes('/prediction/')) {
+        checkDailyReset();
+        
+        const ip = (req.ip || 'unknown').replace(/:\d+[^:]*$/, '');
+        const count = dailyUsage.get(ip) || 0;
+        
+        if (count >= 200) {
+            return res.status(429).json({
+                error: 'Daily limit reached',
+                message: 'You have reached your daily usage limit. Try again tomorrow.'
+            });
+        }
+        
+        dailyUsage.set(ip, count + 1);
+        
+        // FIXED: Prevent unbounded growth
+        if (dailyUsage.size > 10000) {
+            console.warn('[System] Daily usage map too large, clearing oldest entries');
+            const entries = Array.from(dailyUsage.entries()).slice(0, 1000);
+            entries.forEach(([key]) => dailyUsage.delete(key));
+        }
+    }
     next();
 });
 
+// --- BOT DETECTION ---
+app.use((req, res, next) => {
+    // FIXED: Check all POST requests, not just /prediction/
+    if (req.method !== 'POST') {
+        return next();
+    }
+    
+    const userAgent = (req.headers['user-agent'] || '').toLowerCase();
+    const suspiciousBots = ['python-requests', 'curl/', 'wget/', 'scrapy', 'crawler'];
+    
+    const isBot = suspiciousBots.some(bot => userAgent.includes(bot));
+    
+    if (isBot) {
+        console.log(`[Security] Blocked bot: ${userAgent.substring(0, 50)} from ${req.ip}`);
+        return res.status(403).json({
+            error: 'Automated access detected',
+            message: 'This service is for web browsers only.'
+        });
+    }
+    next();
+});
+
+// --- INSTANCES CONFIGURATION ---
 let INSTANCES = [];
 try {
     INSTANCES = JSON.parse(process.env.FLOWISE_INSTANCES || '[]');
     console.log(`[System] Loaded ${INSTANCES.length} instances`);
+    if (!Array.isArray(INSTANCES) || INSTANCES.length === 0) {
+        console.error('ERROR: FLOWISE_INSTANCES must be a non-empty array');
+    }
 } catch (e) {
-    console.error("ERROR parsing FLOWISE_INSTANCES:", e);
+    console.error("CRITICAL ERROR: Could not parse FLOWISE_INSTANCES JSON", e);
 }
 
+// --- CACHE WITH AUTO-CLEANUP ---
 const flowCache = new Map();
 
+setInterval(() => {
+    const now = Date.now();
+    for (const [key, value] of flowCache.entries()) {
+        // FIXED: Safety check for timestamp
+        if (value.timestamp && now - value.timestamp > 10 * 60 * 1000) {
+            flowCache.delete(key);
+        }
+    }
+}, 10 * 60 * 1000);
+
+// --- FETCH WITH TIMEOUT ---
+async function fetchWithTimeout(url, options, timeout = 10000) {
+    return Promise.race([
+        fetch(url, options),
+        new Promise((_, reject) =>
+            setTimeout(() => reject(new Error('Request timeout')), timeout)
+        )
+    ]);
+}
+
+// --- RESOLVE CHATFLOW ID ---
 async function resolveChatflowId(instanceNum, botName) {
     const cacheKey = `${instanceNum}-${botName}`;
+    
     const cached = flowCache.get(cacheKey);
-    if (cached) return cached;
+    if (cached && cached.timestamp && Date.now() - cached.timestamp < 5 * 60 * 1000) {
+        return { id: cached.id, instance: cached.instance };
+    }
     
-    if (instanceNum < 1 || instanceNum > INSTANCES.length) {
-        throw new Error(`Invalid instance: ${instanceNum}`);
+    if (isNaN(instanceNum) || instanceNum < 1 || instanceNum > INSTANCES.length) {
+        throw new Error(`Instance ${instanceNum} does not exist. Valid: 1-${INSTANCES.length}`);
     }
     
     const instance = INSTANCES[instanceNum - 1];
     console.log(`[System] Looking up '${botName}' in instance ${instanceNum}...`);
     
     const headers = {};
-    if (instance.key) headers['Authorization'] = `Bearer ${instance.key}`;
+    if (instance.key && instance.key.length > 0) {
+        headers['Authorization'] = `Bearer ${instance.key}`;
+    }
+    
+    const response = await fetchWithTimeout(`${instance.url}/api/v1/chatflows`, { headers }, 10000);
     
-    const response = await fetch(`${instance.url}/api/v1/chatflows`, { headers });
     if (!response.ok) {
-        throw new Error(`Instance ${instanceNum} returned status ${response.status}. Instance may be paused.`);
+        throw new Error(`Instance ${instanceNum} returned status ${response.status}`);
     }
     
     const flows = await response.json();
-    const match = flows.find(f => f.name.toLowerCase().replace(/\s+/g, '-') === botName);
     
-    if (!match) throw new Error(`Bot '${botName}' not found in instance ${instanceNum}`);
+    if (!Array.isArray(flows)) {
+        throw new Error(`Instance ${instanceNum} returned invalid response`);
+    }
     
-    const result = { id: match.id, instance };
-    flowCache.set(cacheKey, result);
+    const match = flows.find(f => f.name && f.name.toLowerCase().replace(/\s+/g, '-') === botName);
+    
+    if (!match || !match.id) {
+        throw new Error(`Bot '${botName}' not found in instance ${instanceNum}`);
+    }
+    
+    flowCache.set(cacheKey, {
+        id: match.id,
+        instance: instance,
+        timestamp: Date.now()
+    });
     
     console.log(`[System] Found '${botName}' -> ${match.id}`);
-    return result;
+    
+    return { id: match.id, instance };
 }
 
-// IMPROVED: Safe JSON parsing
-async function safeJsonParse(response, url) {
-    const text = await response.text();
+// --- STREAMING HANDLER ---
+async function handleStreamingResponse(flowiseResponse, clientRes) {
+    clientRes.setHeader('Content-Type', 'text/event-stream');
+    clientRes.setHeader('Cache-Control', 'no-cache');
+    clientRes.setHeader('Connection', 'keep-alive');
     
-    try {
-        return JSON.parse(text);
-    } catch (e) {
-        console.error(`[Error] Non-JSON response from ${url}`);
-        console.error(`[Error] Response starts with: ${text.substring(0, 200)}`);
+    console.log('[Streaming] Forwarding SSE stream...');
+    
+    let streamStarted = false;
+    
+    flowiseResponse.body.on('data', (chunk) => {
+        streamStarted = true;
+        clientRes.write(chunk);
+    });
+    
+    flowiseResponse.body.on('end', () => {
+        console.log('[Streaming] Stream completed');
+        clientRes.end();
+    });
+    
+    flowiseResponse.body.on('error', (err) => {
+        console.error('[Streaming Error]', err.message);
         
-        // Return a user-friendly error
-        throw new Error('The Flowise instance returned an invalid response. It may be paused or experiencing errors.');
-    }
+        // FIXED: Send error event if stream already started
+        if (streamStarted) {
+            clientRes.write(`\n\nevent: error\ndata: {"error": "Stream interrupted"}\n\n`);
+        }
+        clientRes.end();
+    });
 }
 
+// --- ROUTE 1: PREDICTION (WITH STREAMING SUPPORT) ---
 app.post('/api/v1/prediction/:instanceNum/:botName', async (req, res) => {
     try {
         const instanceNum = parseInt(req.params.instanceNum);
-        const botName = req.params.botName.toLowerCase();
+        const botName = req.params.botName.toLowerCase().substring(0, 100);
         
-        if (req.body.question && req.body.question.length > 2000) {
-            return res.status(400).json({ error: 'Message too long (max 2000 characters)' });
+        if (!req.body.question || typeof req.body.question !== 'string') {
+            return res.status(400).json({
+                error: 'Invalid request',
+                message: 'Question must be a non-empty string.'
+            });
+        }
+        
+        if (req.body.question.length > 2000) {
+            return res.status(400).json({
+                error: 'Message too long',
+                message: 'Please keep messages under 2000 characters.'
+            });
         }
         
         const { id, instance } = await resolveChatflowId(instanceNum, botName);
         
         const headers = { 'Content-Type': 'application/json' };
-        if (instance.key) headers['Authorization'] = `Bearer ${instance.key}`;
+        if (instance.key && instance.key.length > 0) {
+            headers['Authorization'] = `Bearer ${instance.key}`;
+        }
         
-        const response = await fetch(`${instance.url}/api/v1/prediction/${id}`, {
-            method: 'POST',
-            headers,
-            body: JSON.stringify(req.body)
-        });
+        const response = await fetchWithTimeout(
+            `${instance.url}/api/v1/prediction/${id}`,
+            {
+                method: 'POST',
+                headers,
+                body: JSON.stringify(req.body)
+            },
+            30000
+        );
         
-        // Check if response is OK before parsing
         if (!response.ok) {
             const errorText = await response.text();
             console.error(`[Error] Instance returned ${response.status}: ${errorText.substring(0, 200)}`);
-            return res.status(response.status).json({ 
+            return res.status(response.status).json({
                 error: 'Flowise instance error',
-                message: 'The chatbot instance may be paused or misconfigured. Please check the Flowise instance logs.'
+                message: 'The chatbot instance returned an error.'
             });
         }
         
-        const data = await safeJsonParse(response, instance.url);
-        res.status(200).json(data);
+        const contentType = response.headers.get('content-type') || '';
+        
+        // STREAMING RESPONSE
+        if (contentType.includes('text/event-stream')) {
+            console.log('[Streaming] Detected SSE response');
+            return handleStreamingResponse(response, res);
+        }
+        
+        // NON-STREAMING RESPONSE
+        console.log('[Non-streaming] Parsing JSON response');
+        const text = await response.text();
+        
+        try {
+            const data = JSON.parse(text);
+            res.status(200).json(data);
+        } catch (e) {
+            console.error('[Error] Invalid JSON:', text.substring(0, 200));
+            res.status(500).json({ error: 'Invalid response from Flowise' });
+        }
         
     } catch (error) {
         console.error('[Error]', error.message);
-        res.status(500).json({ 
+        res.status(500).json({
             error: 'Request failed',
-            message: error.message 
+            message: error.message
         });
     }
 });
 
+// --- ROUTE 2: CHATBOT CONFIG ---
 app.get('/api/v1/public-chatbotConfig/:instanceNum/:botName', async (req, res) => {
     try {
         const instanceNum = parseInt(req.params.instanceNum);
-        const botName = req.params.botName.toLowerCase();
+        const botName = req.params.botName.toLowerCase().substring(0, 100);
+        
         const { id, instance } = await resolveChatflowId(instanceNum, botName);
         
         const headers = {};
-        if (instance.key) headers['Authorization'] = `Bearer ${instance.key}`;
+        if (instance.key && instance.key.length > 0) {
+            headers['Authorization'] = `Bearer ${instance.key}`;
+        }
         
-        const response = await fetch(`${instance.url}/api/v1/public-chatbotConfig/${id}`, { headers });
+        const response = await fetchWithTimeout(
+            `${instance.url}/api/v1/public-chatbotConfig/${id}`,
+            { headers },
+            10000
+        );
         
         if (!response.ok) {
             return res.status(response.status).json({ error: 'Config not available' });
         }
         
-        const data = await safeJsonParse(response, instance.url);
+        const data = await response.json();
         res.status(200).json(data);
+        
     } catch (error) {
         console.error('[Error Config]', error.message);
         res.status(404).json({ error: error.message });
     }
 });
 
+// --- ROUTE 3: STREAMING CHECK ---
 app.get('/api/v1/chatflows-streaming/:instanceNum/:botName', async (req, res) => {
     try {
         const instanceNum = parseInt(req.params.instanceNum);
-        const botName = req.params.botName.toLowerCase();
+        const botName = req.params.botName.toLowerCase().substring(0, 100);
+        
         const { id, instance } = await resolveChatflowId(instanceNum, botName);
         
         const headers = {};
-        if (instance.key) headers['Authorization'] = `Bearer ${instance.key}`;
+        if (instance.key && instance.key.length > 0) {
+            headers['Authorization'] = `Bearer ${instance.key}`;
+        }
         
-        const response = await fetch(`${instance.url}/api/v1/chatflows-streaming/${id}`, { headers });
+        const response = await fetchWithTimeout(
+            `${instance.url}/api/v1/chatflows-streaming/${id}`,
+            { headers },
+            10000
+        );
         
         if (!response.ok) {
-            return res.status(response.status).json({ isStreaming: false });
+            return res.status(200).json({ isStreaming: false });
         }
         
-        const data = await safeJsonParse(response, instance.url);
+        const data = await response.json();
         res.status(200).json(data);
+        
     } catch (error) {
         console.error('[Error Streaming]', error.message);
-        res.status(200).json({ isStreaming: false }); // Default to non-streaming
+        res.status(200).json({ isStreaming: false });
     }
 });
 
+// --- HEALTH CHECK ---
 app.get('/', (req, res) => res.send('Federated Proxy Active'));
-app.get('/health', (req, res) => res.json({ status: 'ok', instances: INSTANCES.length }));
 
-app.listen(7860, '0.0.0.0', () => console.log('Federated Proxy running on port 7860'));
+app.get('/health', (req, res) => {
+    res.json({
+        status: 'healthy',
+        instances: INSTANCES.length,
+        cached_bots: flowCache.size,
+        daily_active_ips: dailyUsage.size,
+        uptime: process.uptime()
+    });
+});
+
+// --- 404 HANDLER ---
+app.use((req, res) => {
+    res.status(404).json({ error: 'Route not found' });
+});
+
+// --- GLOBAL ERROR HANDLER ---
+app.use((err, req, res, next) => {
+    console.error('[Error] Unhandled error:', err);
+    res.status(500).json({ error: 'Internal server error' });
+});
+
+// --- GRACEFUL SHUTDOWN ---
+const server = app.listen(7860, '0.0.0.0', () => {
+    console.log('Federated Proxy running on port 7860');
+});
+
+process.on('SIGTERM', () => {
+    console.log('[System] SIGTERM received, shutting down gracefully...');
+    server.close(() => {
+        console.log('[System] Server closed');
+        process.exit(0);
+    });
+});
+
+process.on('SIGINT', () => {
+    console.log('[System] SIGINT received, shutting down gracefully...');
+    server.close(() => {
+        console.log('[System] Server closed');
+        process.exit(0);
+    });
+});