ServerEngine.cpp

#include "ServerEngine.hpp"
#include "Utils.hpp"
#include "ResponseBuilder.hpp"
#include <iostream>

ServerEngine::ServerEngine() {}
ServerEngine::~ServerEngine() {
    // Clean up clients
    for (std::vector<Client*>::iterator it = clients.begin(); it != clients.end(); ++it) {
        delete *it;
    }
    clients.clear();
}

bool ServerEngine::isServerFD(int fd) const {
    for (std::map<int, size_t>::const_iterator it = fd_to_server_index.begin();
         it != fd_to_server_index.end(); ++it) {
        if (it->first == fd && servers[it->second].getServerFD() == fd) {
            return true;
        }
    }
    return false;
}

bool compressServerBlocks(std::vector<Config> &configs, std::vector<ServerConfig> &_server_config)
{
    if (configs.empty()) {
        debug("No configurations to process");
        return false;
    }

    // Use a map to track unique IP:Port combinations
    std::map<std::pair<std::string, int>, ServerConfig*> serverMap;

    for (std::vector<Config>::const_iterator config_it = configs.begin();
         config_it != configs.end(); config_it++)
    {
        std::string temp_ip = config_it->get_ip();
        int temp_port = config_it->get_port();
        std::string server_name = config_it->get_server_name();

        // Create a unique key for IP:Port
        std::pair<std::string, int> key = std::make_pair(temp_ip, temp_port);

        // Find or create ServerConfig
        ServerConfig* currentServerConfig = NULL;

        // Search for existing configuration
        std::map<std::pair<std::string, int>, ServerConfig*>::iterator it = serverMap.find(key);
        if (it == serverMap.end()) {
            // Create a new server configuration
            _server_config.push_back(ServerConfig(temp_ip, temp_port));
            currentServerConfig = &_server_config.back();
            serverMap[key] = currentServerConfig;
        } else {
            currentServerConfig = it->second;
        }

        // Attempt to add configuration
        if (!currentServerConfig->addConfig(*config_it)) {
            // Handle configuration conflict
            debug("Conflict adding configuration for: " + server_name);

            // Conflict resolution strategy
            // For example, use the latest configuration or ignore
            continue;
        }
    }

    debug("Configurations successfully compressed. Total: " +
           Utils::to_string(_server_config.size()) + " server blocks");

    return true;
}

bool ServerEngine::loadConfig(const std::string& configFile) {

    if (!_parser.parseFile(configFile, configs)) {
        std::cerr << "Error parsing config file: " << configFile << std::endl;
        return false;
    }
    if (!compressServerBlocks(configs, _server_config)){
        std::cerr << "Error compressing Config to ServerConfig: " << std::endl;
        return (false);
    }
    return true;
}

size_t ServerEngine::getTotalActiveClients() const {
    size_t total = 0;
    std::vector<Server>::const_iterator server_it;
    for (server_it = servers.begin(); server_it != servers.end(); ++server_it) {
        total += server_it->getActiveClients();
    }
    return total;
}

size_t ServerEngine::getTotalConnections() const {
    size_t total = 0;
    std::vector<Server>::const_iterator server_it;
    for (server_it = servers.begin(); server_it != servers.end(); ++server_it) {
        total += server_it->getTotalConnections();
    }
    return total;
}

void ServerEngine::printGlobalStats() const {
    std::cout << "\n====== Global Server Statistics ======\n"
              << "Total servers running: " << getTotalServers() << "\n"
              << "Total active clients: " << getTotalActiveClients() << "\n"
              << "Total historical connections: " << getTotalConnections() << "\n\n";

    std::cout << "Individual Server Stats:\n";
    std::vector<Server>::const_iterator server_it;
    for (server_it = servers.begin(); server_it != servers.end(); ++server_it) {
        server_it->printServerStats();
    }
    std::cout << "==================================\n";
}

bool ServerEngine::start_func_test(const std::string& configFile) {
    if (!loadConfig(configFile)) {
        return false;
    }

    // Create servers for each ServerConfig
    for (std::vector<Config>::const_iterator config_it = configs.begin();
         config_it != configs.end(); ++config_it) {
        servers.push_back(Server(*config_it));
    }

    return true;
}

int ServerEngine::start(const std::string& configFile) {
    if (!loadConfig(configFile)) {
        return -1;
    }

    std::cout << "Total ServerConfigs: " << _server_config.size() << std::endl;

    // Map to track sockets by IP:port combination -> Moved to ServerEngine.hpp
    // std::map<std::pair<std::string, int>, size_t> ipPortToServerIndex;

    // First pass: Create servers and set up sockets
    for (std::vector<ServerConfig>::const_iterator config_it = _server_config.begin();
         config_it != _server_config.end(); ++config_it) {
        std::vector<Config> server_configs = config_it->get_Configs();

        std::cout << "Current ServerConfig: IP=" << config_it->get_Ip()
                  << ", Port=" << config_it->get_Port()
                  << ", Config Count=" << server_configs.size() << std::endl;

        if (server_configs.empty()) {
            continue;
        }

        // Key for the IP:port map
        std::pair<std::string, int> port_key =
            std::make_pair(config_it->get_Ip(), config_it->get_Port());

        // Check if a server already exists for this IP:port combination
        if (ipPortToServerIndex.find(port_key) == ipPortToServerIndex.end()) {
            // First server for this IP:port combination
            std::cout << "  Creating server for IP=" << config_it->get_Ip()
                    << ", Port=" << config_it->get_Port()
                    << " with " << server_configs.size() << " virtual hosts" << std::endl;

            // Create server with all the ServerConfig configuration
            servers.push_back(Server(*config_it));
            size_t server_index = servers.size() - 1;

            // Try to set up the socket
            if (!servers[server_index].setup()) {
                std::cerr << "Failed to setup server for "
                        << config_it->get_Ip() << ":" << config_it->get_Port()
                        << std::endl;
                // Remove the server that failed
                servers.pop_back();
                continue;
            }

            // Socket configured correctly
            int server_fd = servers[server_index].getServerFD();
            if (server_fd >= 0) {
                pollfd server_pollfd = {server_fd, POLLIN, 0};
                poll_fds.push_back(server_pollfd);

                // Store the server index
                fd_to_server_index[server_fd] = server_index;
                ipPortToServerIndex[port_key] = server_index;
            } else {
                // Invalid socket, remove the server
                std::cerr << "Warning: Server created but socket is invalid for "
                          << config_it->get_Ip() << ":" << config_it->get_Port()
                          << std::endl;
                servers.pop_back();
            }
        } else {
            // A server already exists for this IP:port
            // Add additional configurations to the existing server
            size_t existing_index = ipPortToServerIndex[port_key];
            std::cout << "  Adding virtual hosts to existing server for IP="
                      << config_it->get_Ip() << ", Port=" << config_it->get_Port()
                      << std::endl;

            // Get the current server configuration
            ServerConfig existingConfig = servers[existing_index].getServerConfig();

            // Add all configurations from the current ServerConfig
            for (std::vector<Config>::const_iterator conf_it = server_configs.begin();
                 conf_it != server_configs.end(); ++conf_it) {
                existingConfig.addConfig(*conf_it);
                std::cout << "  Added server_name=" << conf_it->get_server_name()
                          << " to existing server" << std::endl;
            }

            // Create a new server with the combined configuration
            // and with the same socket_fd as the existing server
            int existing_fd = servers[existing_index].getServerFD();
            servers[existing_index].updateWithServerConfig(existingConfig, existing_fd);
            // IMPORTANT: Restore the server_fd to the original value
            // Since the Server constructor sets it to -1
            servers[existing_index].server_fd = existing_fd;
        }
    }

    std::cout << "Total Servers created: " << servers.size() << std::endl;

    if (!servers.empty()) {
        for (std::vector<Server>::iterator i = servers.begin();
            i != servers.end(); i++)
            debug(i->getServerConfig());
    }

    //printDebugInfo();
    printGlobalStats();
    pollLoop();

    return 0;
}

struct InvalidPollFD {
    bool operator()(const pollfd& pfd) const {
        return pfd.fd == -1;
    }
};

void ServerEngine::pollLoop() {
    debug("[ServerEngine]: Starting poll loop");

    const int POLL_TIMEOUT = 5000;        // 5 seconds timeout
    const int TIMEOUT_CHECK_INTERVAL = 30; // Check timeouts every 30 iterations
    //const int MAX_BACKLOG_SIZE = 128;     // Maximum number of pending connections
    const int MAX_CLIENT_COUNT = 2048;    // Maximum number of clients to accept
    const int CLIENT_TIMEOUT = 6000;        // Client inactivity timeout in seconds
    const int UPLOAD_TIMEOUT = 300;       // Upload timeout in seconds (5 minutes)

    int iteration_count = 0;
    time_t last_stats_time = time(NULL);
    bool server_overloaded = false;

    while (true) {
        try {
            // Check if we have too many clients and need to temporarily stop accepting
            size_t total_clients = getTotalActiveClients();
            if (total_clients > MAX_CLIENT_COUNT) {
                if (!server_overloaded) {
                    // debug("[ServerEngine]: Server overloaded, temporarily stopping new connections. " +
                    //       "Active clients: " + Utils::to_string(total_clients));
                    server_overloaded = true;
                }
            } else if (server_overloaded && total_clients < MAX_CLIENT_COUNT * 0.8) {
                // Resume accepting connections when load drops below 80%
                // debug("[ServerEngine]: Server load returning to normal, resuming new connections. " +
                //       "Active clients: " + Utils::to_string(total_clients));
                server_overloaded = false;
            }

            // Update poll events based on client states
            for (std::vector<Server>::iterator server_it = servers.begin(); 
                 server_it != servers.end(); ++server_it) {
                server_it->getConnectionHandler().updatePollEvents(poll_fds);
            }

            // Poll for events
            int poll_count = poll(poll_fds.data(), poll_fds.size(), POLL_TIMEOUT);

            if (poll_count < 0) {
                if (errno == EINTR) {
                    debug("[ServerEngine]: Poll interrupted by signal, continuing");
                    continue;
                }

                debug("[ServerEngine]: Poll error: " + Utils::to_string(errno) + " (" + strerror(errno) + ")");
                // For recoverable errors, continue
                if (errno == EAGAIN || errno == EWOULDBLOCK) {
                    continue;
                }
                
                // For more serious errors, pause briefly to avoid CPU spinning
                debug("[ServerEngine]: Serious poll error, pausing briefly");
                usleep(100000); // 100ms pause
                continue;
            }

            if (poll_count == 0) {
                debug("[ServerEngine]: Poll timeout, checking for stuck uploads and inactive clients");
                
                // Check for stuck uploads that are almost complete and force completion
                for (std::vector<pollfd>::iterator pfd_it = poll_fds.begin(); 
                     pfd_it != poll_fds.end(); ++pfd_it) {
                    
                    int client_fd = pfd_it->fd;
                    
                    // Skip server fds
                    if (isServerFD(client_fd)) {
                        continue;
                    }
                    
                    // Find the server for this client
                    std::map<int, size_t>::iterator server_it = fd_to_server_index.find(client_fd);
                    if (server_it == fd_to_server_index.end()) {
                        continue;
                    }
                    
                    Server& server = servers[server_it->second];
                    ConnectionHandler& connHandler = server.getConnectionHandler();
                    Client* client = connHandler.getClient(client_fd);
                    
                    if (!client) {
                        continue;
                    }
                    
                    // Check for stuck multipart uploads
                    if (client->getState() == STATE_READING_REQUEST && 
                        client->isMultiPartRequest() && !client->isBodyComplete()) {
                        
                        // Get expected content length
                        int contentLength = 0;
                        if (client->getCurrentRequest()) {
                            std::string clStr = client->getCurrentRequest()->getHeaders().get("Content-Length");
                            if (!clStr.empty()) {
                                contentLength = std::atoi(clStr.c_str());
                            }
                        }
                        
                        // Calculate how complete the upload is and how long since last activity
                        double completionPercentage = (contentLength > 0) ? 
                            static_cast<double>(client->getTotalBody().size()) / contentLength : 0;
                        time_t idleTime = time(NULL) - client->getLastActivityTime();
                        
                        // If we have 98% or more of the data and we haven't received more in 5 seconds
                        if (contentLength > 0 && completionPercentage >= 0.98 && idleTime > 5) {
                            debug("[ServerEngine]: Upload almost complete (" + 
                                  Utils::to_string(completionPercentage * 100) + "%) and stuck for " +
                                  Utils::to_string(idleTime) + " seconds, forcing completion");
                            
                            client->forceCompleteMultipartUpload();
                            
                            // Process the request
                            server.processRequest(client);
                            continue;
                        }
                        
                        // Check for upload timeout
                        if (idleTime > UPLOAD_TIMEOUT) {
                            debug("[ServerEngine]: Upload timed out after " + Utils::to_string(idleTime) + 
                                  " seconds, aborting");
                            
                            // Create timeout response
                            responseBuilder.reset();
                            responseBuilder.setStatus(408, "Request Timeout");
                            responseBuilder.setContentType("text/plain");
                            responseBuilder.setBody("Upload timed out after " + 
                                                  Utils::to_string(UPLOAD_TIMEOUT) + " seconds");
                            
                            std::string response = responseBuilder.build();
                            client->setResponseBuffer(response);
                            continue;
                        }
                    }
                    
                    // Check for general client timeout
                    time_t idleTime = time(NULL) - client->getLastActivityTime();
                    if (idleTime > CLIENT_TIMEOUT) {
                        debug("[ServerEngine]: Client " + Utils::to_string(client_fd) + 
                              " inactive for " + Utils::to_string(idleTime) + " seconds, closing");
                        
                        connHandler.closeClient(client_fd);
                        
                        // Mark for removal from poll_fds
                        pfd_it->fd = -1;
                    }
                }
                
                // Remove any marked fds
                poll_fds.erase(
                    std::remove_if(poll_fds.begin(), poll_fds.end(), InvalidPollFD()),
                    poll_fds.end()
                );
                
                continue;
            }

            // List of file descriptors to remove
            std::vector<int> fds_to_remove;

            // Process events
            for (size_t i = 0; i < poll_fds.size(); ++i) {
                pollfd& pfd = poll_fds[i];

                if (!pfd.revents) {
                    continue; // No events for this FD
                }

                // Check if this is a server socket
                bool isServer = isServerFD(pfd.fd);

                // Handle incoming connections on server sockets
                if (isServer && (pfd.revents & POLLIN)) {
                    // Skip accepting new connections if server is overloaded
                    if (server_overloaded) {
                        debug("[ServerEngine]: Server overloaded, ignoring new connection attempt");
                        continue;
                    }
                    
                    debug("[ServerEngine]: New client connecting to server FD " + Utils::to_string(pfd.fd));
                    size_t server_idx = fd_to_server_index[pfd.fd];
                    servers[server_idx].acceptNewClient(poll_fds, fd_to_server_index, servers);
                    continue;
                }

                // Handle client events - separate read/write events from error events
                if (!isServer) {
                    std::map<int, size_t>::iterator server_it = fd_to_server_index.find(pfd.fd);
                    if (server_it == fd_to_server_index.end()) {
                        debug("[ServerEngine]: FD not found in fd_to_server_index: " + Utils::to_string(pfd.fd));
                        fds_to_remove.push_back(pfd.fd);
                        continue;
                    }

                    Server& server = servers[server_it->second];
                    ConnectionHandler& connHandler = server.getConnectionHandler();
                    Client* client = connHandler.getClient(pfd.fd);

                    if (!client) {
                        debug("[ServerEngine]: Client object not found for fd " + Utils::to_string(pfd.fd));
                        fds_to_remove.push_back(pfd.fd);
                        continue;
                    }

                    // Update last activity time
                    client->updateLastActivityTime();
                    
                    // Handle data events first, then check for errors
                    bool handled = false;

                    // Handle read events for clients in reading or idle state
                    if ((pfd.revents & POLLIN) && 
                        (client->getState() == STATE_READING_REQUEST || client->getState() == STATE_IDLE)) {
                        
                        debug("[ServerEngine]: Read event for client fd " + Utils::to_string(pfd.fd) + 
                              " in state " + client->stateToString(client->getState()));
                        
                        // Try to read data from client
                        int result = client->receiveRequest();
                        handled = true;
                        
                        if (result < 0) {
                            // Error reading, close client
                            debug("[ServerEngine]: Error receiving from client, removing");
                            connHandler.closeClient(pfd.fd);
                            fds_to_remove.push_back(pfd.fd);
                            continue;
                        }
                        
                        // Process completed requests
                        if (result == 1 && client->getState() == STATE_PROCESSING) {
                            debug("[ServerEngine]: Request complete, processing");
                            
                            try {
                                server.processRequest(client);
                                debug("[ServerEngine]: Request processed, client state now: " + 
                                     client->stateToString(client->getState()));
                                
                                // Update poll events based on new state
                                if (client->getState() == STATE_SENDING_RESPONSE) {
                                    debug("[ServerEngine]: Switching client to write mode");
                                    pfd.events = POLLOUT; // Switch to write events
                                }
                                
                            } catch (const std::exception& e) {
                                debug("[ServerEngine]: Exception in processRequest: " + std::string(e.what()));
                                
                                // Try to send an error response
                                try {
                                    responseBuilder.reset();
                                    responseBuilder.setStatus(500, "Internal Server Error");
                                    responseBuilder.setContentType("text/plain");
                                    responseBuilder.setBody("Internal server error: " + std::string(e.what()));
                                    std::string response = responseBuilder.build();
                                    
                                    client->setResponseBuffer(response);
                                    client->setState(STATE_SENDING_RESPONSE);
                                    pfd.events = POLLOUT; // Switch to write events
                                } catch (...) {
                                    // If that fails too, just close the connection
                                    connHandler.closeClient(pfd.fd);
                                    fds_to_remove.push_back(pfd.fd);
                                }
                                continue;
                            }
                        }
                    }
                    
                    // Handle write events for clients in sending state
                    if ((pfd.revents & POLLOUT) && client->getState() == STATE_SENDING_RESPONSE) {
                        debug("[ServerEngine]: Write event for client fd " + Utils::to_string(pfd.fd));
                        
                        int result = client->sendResponse();
                        handled = true;
                        
                        if (result < 0) {
                            debug("[ServerEngine]: Error sending to client, removing");
                            connHandler.closeClient(pfd.fd);
                            fds_to_remove.push_back(pfd.fd);
                            continue;
                        }
                        
                        if (result > 0) {
                            debug("[ServerEngine]: Response sent, closing non-keep-alive connection");
                            connHandler.closeClient(pfd.fd);
                            fds_to_remove.push_back(pfd.fd);
                            continue;
                        }
                        
                        // Keep-alive connections should go back to reading mode
                        if (client->getState() == STATE_IDLE) {
                            debug("[ServerEngine]: Keep-alive connection reset to IDLE state, switching to read mode");
                            pfd.events = POLLIN;  // Change back to read events
                        }
                    }

                    // Handle error/disconnection events
                    if (pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) {
                        debug("[ServerEngine]: Error or disconnection on fd: " + 
                              Utils::to_string(pfd.fd) + ", events: " + Utils::to_string(pfd.revents));
                        
                        // Only close if we haven't already handled this client
                        if (!handled) {
                            connHandler.closeClient(pfd.fd);
                            fds_to_remove.push_back(pfd.fd);
                        }
                        continue;
                    }
                    
                    // Set appropriate poll events for next iteration
                    if (client->getState() == STATE_SENDING_RESPONSE) {
                        pfd.events = POLLOUT;
                    } else {
                        pfd.events = POLLIN;
                    }
                }
            }

            // Safe removal of file descriptors
            for (std::vector<int>::iterator fd_it = fds_to_remove.begin();
                 fd_it != fds_to_remove.end(); ++fd_it) {

                debug("[ServerEngine]: Removing fd " + Utils::to_string(*fd_it) + " from tracking");
                
                // Remove from fd_to_server_index
                fd_to_server_index.erase(*fd_it);

                // Remove from poll_fds
                for (std::vector<pollfd>::iterator pfd_it = poll_fds.begin();
                     pfd_it != poll_fds.end(); ) {
                    if (pfd_it->fd == *fd_it) {
                        pfd_it = poll_fds.erase(pfd_it);
                    } else {
                        ++pfd_it;
                    }
                }
            }

            // Periodic checks for timeouts and statistics
            if (++iteration_count % TIMEOUT_CHECK_INTERVAL == 0) {
                debug("[ServerEngine]: Checking inactive connections after " + 
                      Utils::to_string(iteration_count) + " iterations");
                
                // Only print stats every 60 seconds
                time_t now = time(NULL);
                if (now - last_stats_time >= 60) {
                    printGlobalStats();
                    last_stats_time = now;
                }
                
                // Check inactive clients for each server
                for (std::vector<Server>::iterator server_it = servers.begin(); 
                     server_it != servers.end(); ++server_it) {
                    server_it->getConnectionHandler().closeInactiveClients(poll_fds, CLIENT_TIMEOUT);
                }
                
                // Periodic garbage collection - remove any closed fds from poll_fds
                for (std::vector<pollfd>::iterator pfd_it = poll_fds.begin();
                     pfd_it != poll_fds.end(); ) {
                    
                    // Check if the fd is valid
                    if (pfd_it->fd >= 0 && fcntl(pfd_it->fd, F_GETFD) == -1) {
                        // Invalid fd, remove from polling
                        debug("[ServerEngine]: Removing invalid fd " + Utils::to_string(pfd_it->fd) + " from poll_fds");
                        fd_to_server_index.erase(pfd_it->fd);
                        pfd_it = poll_fds.erase(pfd_it);
                    } else {
                        ++pfd_it;
                    }
                }
            }
        } catch (const std::exception& e) {
            debug("[ServerEngine]: Exception in poll loop: " + std::string(e.what()));
            // Pause briefly to avoid CPU spinning on repeated exceptions
            usleep(100000); // 100ms pause
        } catch (...) {
            debug("[ServerEngine]: Unknown exception in poll loop");
            usleep(100000); // 100ms pause
        }
    }
}

Client* ServerEngine::findClient(int client_fd) {
    for (std::vector<Client*>::iterator it = clients.begin(); it != clients.end(); ++it) {
        if ((*it)->getClientFD() == client_fd) {
            return *it;
        }
    }
    return NULL;
}

void ServerEngine::removeClient(int client_fd) {
    for (std::vector<Client*>::iterator it = clients.begin(); it != clients.end(); ++it) {
        if ((*it)->getClientFD() == client_fd) {
            debug("[ServerEngine]: Removing client with fd " + Utils::to_string(client_fd));
            delete *it;
            clients.erase(it);
            break;
        }
    }
}