cards/src/server.cpp

#include "server.hpp"
#include <boost/asio/detail/chrono.hpp>
#include <iostream>
#include <chrono>

HttpSession::HttpSession(net::ip::tcp::socket&& socket,
                         std::shared_ptr<Database> db,
                         std::shared_ptr<std::atomic<std::size_t>> sessionCounter)
    : stream(std::move(socket)), db(db), sessionCounter(sessionCounter) {
    if (sessionCounter) {
        sessionCounter->fetch_add(1);
    }
}

HttpSession::~HttpSession() {
    if (sessionCounter) {
        sessionCounter->fetch_sub(1);
    }
}

void HttpSession::run() {
    readRequest();
}

void HttpSession::readRequest() {
    req = {};
    stream.expires_after(net::chrono::seconds(15));
    http::async_read(
        stream,
        buffer,
        req,
        beast::bind_front_handler(
            &HttpSession::onRead,
            shared_from_this()
        )
    );
}

void HttpSession::onRead(beast::error_code ec, std::size_t) {
    if (ec) {
        handleError(ec, "read");
        return;
    }
    processRequest(std::move(req));
}

void HttpSession::processRequest(http::request<http::string_body>&& request) {
    auto self = shared_from_this();

    auto sendJson = [self](http::status status, unsigned version, json body) {
        http::response<http::string_body> res{status, version};
        res.set(http::field::server, "Cards Server");
        res.set(http::field::content_type, "application/json");
        res.body() = body.dump(4);
        res.prepare_payload();
        self->writeResponse(std::move(res));
    };

    if (request.method() == http::verb::get && request.target() == "/") {
        db->getRandomRecord([self, sendJson, version = request.version()](json body) {
            net::post(
                self->stream.get_executor(),
                [self, sendJson, version, body = std::move(body)]() mutable {
                    sendJson(http::status::ok, version, std::move(body));
                });
        });
        return;
    }

    if (request.method() == http::verb::get && request.target().starts_with("/cards")) {
        int count = 0;
        try {
            std::string number{request.target()};
            auto pos = number.find("?count=");
            if (pos == std::string::npos) {
                throw std::invalid_argument("missing count");
            }
            size_t start = pos + 7;
            size_t end = number.find('?', start);
            if (end == std::string::npos) {
                end = number.length();
            }
            count = std::stoi(number.substr(start, end - start));
            count = std::max(1, std::min(count, 1000));
        } catch (const std::exception&) {
            sendJson(http::status::bad_request, request.version(),
                     json{{"error", "Invalid count parameter"}});
            return;
        }
        db->getRandomRecords(count, 10, 1000,
            [self, sendJson, version = request.version()](json body) {
                net::post(
                    self->stream.get_executor(),
                    [self, sendJson, version, body = std::move(body)]() mutable {
                        sendJson(http::status::ok, version, std::move(body));
                    });
            });
        return;
    }

    if (request.method() == http::verb::get && request.target() == "/health") {
        sendJson(http::status::ok, request.version(), json{
            {"status", "ok"},
            {"timestamp", std::chrono::system_clock::now().time_since_epoch().count()}
        });
        return;
    }

    sendJson(http::status::not_found, request.version(), json{
        {"error", "Not Found"}
    });
}

void HttpSession::writeResponse(http::response<http::string_body> res) {
    auto self = shared_from_this();
    auto response = std::make_shared<http::response<http::string_body>>(std::move(res));
    http::async_write(
        stream,
        *response,
        [self, response](beast::error_code ec, std::size_t) {
            if (ec) {
                self->handleError(ec, "write");
            }
            beast::error_code ignored;
            self->stream.socket().shutdown(net::ip::tcp::socket::shutdown_send, ignored);
        }
    );
}

void HttpSession::handleError(beast::error_code ec, const char* what) {
    if (ec == net::error::eof || ec == net::error::connection_reset) {
        // Normal disconnection
        return;
    }
    std::cerr << "Error in " << what << ": " << ec.message() << std::endl;
}

// Implementation of Server
Server::Server(const std::string& address, int port, int threads, std::shared_ptr<Database> db)
    : ioc(threads),
      acceptor(ioc, net::ip::tcp::endpoint(net::ip::make_address(address), port), true),
      db(db),
      threads(threads),
      acceptTimer(ioc),
      activeSessions(std::make_shared<std::atomic<std::size_t>>(0)) {}

void Server::run() {
    accept();

    // Run the I/O service on the requested number of threads
    threadPool.reserve(threads);
    for (int i = 0; i < threads; ++i) {
        threadPool.emplace_back([this]() {
            ioc.run();
        });
    }

    // Wait for all threads to finish
    for (auto& t : threadPool) {
        if (t.joinable()) {
            t.join();
        }
    }
}

void Server::stop() {
    ioc.stop();
}

void Server::accept() {
    acceptor.async_accept(
        net::make_strand(ioc),
        beast::bind_front_handler(&Server::onAccept, this)
    );
}

void Server::onAccept(beast::error_code ec, net::ip::tcp::socket socket) {
    if (ec) {
        std::cerr << "Accept error: " << ec.message() << std::endl;
        if (!ioc.stopped()) {
            acceptTimer.expires_after(std::chrono::milliseconds(100));
            acceptTimer.async_wait([this](beast::error_code ec) {
                if (!ioc.stopped()) { accept(); }
            });
        }
        return;
    }

    if (activeSessions->load() >= maxSessions) {
        beast::error_code ignore;
        socket.close(ignore);
    } else {
        std::make_shared<HttpSession>(std::move(socket), db, activeSessions)->run();
    }
    accept();
}