server_session.hpp

/* -*- mode: c++; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*-
 *
 * Copyright (C) 2025 Marco Craveiro <marco.craveiro@gmail.com>
 *
 * This program is free software; you can redistribute it and/or modify it under
 * the terms of the GNU General Public License as published by the Free Software
 * Foundation; either version 3 of the License, or (at your option) any later
 * version.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
 * FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
 * details.
 *
 * You should have received a copy of the GNU General Public License along with
 * this program; if not, write to the Free Software Foundation, Inc., 51
 * Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
 *
 */
#ifndef ORES_COMMS_NET_SERVER_SESSION_HPP
#define ORES_COMMS_NET_SERVER_SESSION_HPP
 
#include <mutex>
#include <queue>
#include <atomic>
#include <memory>
#include <string>
#include <chrono>
#include <boost/asio/steady_timer.hpp>
#include <boost/asio/strand.hpp>
#include "ores.comms/net/connection.hpp"
#include "ores.logging/make_logger.hpp"
#include "ores.comms/messaging/message_dispatcher.hpp"
#include "ores.comms/service/auth_session_service.hpp"
 
namespace ores::comms::service { class subscription_manager; }
 
namespace ores::comms::net {
 
class server_session final {
private:
    inline static std::string_view logger_name = "ores.comms.net.server_session";
 
    static auto& lg() {
        using namespace ores::logging;
        static auto instance = make_logger(logger_name);
        return instance;
    }
 
public:
    struct pending_notification {
        std::string event_type;
        std::chrono::system_clock::time_point timestamp;
        std::vector<std::string> entity_ids;
    };
 
    struct pending_database_status {
        bool available;
        std::string error_message;
        std::chrono::system_clock::time_point timestamp;
    };
 
    explicit server_session(std::unique_ptr<connection> conn, std::string server_id,
        std::shared_ptr<messaging::message_dispatcher> dispatcher,
        boost::asio::any_io_executor io_executor,
        std::shared_ptr<service::auth_session_service> sessions,
        std::shared_ptr<service::subscription_manager> subscription_mgr = nullptr);
 
    boost::asio::awaitable<void> run();
 
    void stop();
 
    bool queue_notification(const std::string& event_type,
        std::chrono::system_clock::time_point timestamp,
        const std::vector<std::string>& entity_ids = {});
 
    bool queue_database_status(bool available, const std::string& error_message,
        std::chrono::system_clock::time_point timestamp);
 
private:
    boost::asio::awaitable<bool> perform_handshake();
 
    boost::asio::awaitable<void> process_messages();
 
    boost::asio::awaitable<void> run_notification_writer();
 
    boost::asio::awaitable<void> send_pending_notifications();
 
    boost::asio::awaitable<void> send_pending_database_status();
 
    boost::asio::awaitable<void> write_frame_serialized(const messaging::frame& frame);
 
    void register_with_subscription_manager();
 
    void unregister_from_subscription_manager();
 
    std::unique_ptr<connection> conn_;
    std::string server_id_;
    std::shared_ptr<messaging::message_dispatcher> dispatcher_;
    std::shared_ptr<service::auth_session_service> sessions_;
    std::shared_ptr<service::subscription_manager> subscription_mgr_;
    std::uint32_t sequence_number_;
    bool handshake_complete_;
    std::atomic<bool> active_{false};
 
    // Strand for serializing write operations
    boost::asio::strand<boost::asio::any_io_executor> write_strand_;
 
    // Atomic flag to serialize write operations (prevents concurrent SSL writes)
    std::atomic<bool> write_in_progress_{false};
 
    // Timer used as async signal for notification availability
    boost::asio::steady_timer notification_signal_;
 
    // Thread-safe notification queue
    mutable std::mutex notification_mutex_;
    std::queue<pending_notification> pending_notifications_;
    std::queue<pending_database_status> pending_database_status_;
 
    // Session compression type negotiated during handshake
    messaging::compression_type session_compression_{messaging::compression_type::none};
};
 
}
 
#endif