ClientManager.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_QT_CLIENT_MANAGER_HPP
#define ORES_QT_CLIENT_MANAGER_HPP
 
#include <atomic>
#include <chrono>
#include <concepts>
#include <expected>
#include <memory>
#include <optional>
#include <string>
#include <filesystem>
#include <boost/uuid/uuid.hpp>
#include <boost/uuid/uuid_io.hpp>
#include <boost/uuid/random_generator.hpp>
#include <rfl/json.hpp>
#include <rfl/AddTagsToVariants.hpp>
#include <QObject>
#include <QTimer>
#include "ores.utility/rfl/reflectors.hpp"
#include <QDateTime>
#include "ores.nats/service/nats_client.hpp"
#include "ores.nats/service/nats_connect_error.hpp"
#include "ores.nats/service/session_expired_error.hpp"
#include "ores.nats/service/jetstream_admin.hpp"
#include "ores.nats/service/subscription.hpp"
#include "ores.eventing/service/event_bus.hpp"
#include "ores.logging/make_logger.hpp"
#include "ores.iam.api/domain/session.hpp"
#include "ores.iam.api/messaging/session_samples_protocol.hpp"
#include "ores.trading.api/messaging/trade_protocol.hpp"
#include "ores.qt/export.hpp"
 
namespace ores::qt {
 
template <typename T>
concept nats_request = requires {
    { T::nats_subject } -> std::convertible_to<std::string_view>;
    typename T::response_type;
};
 
struct PartyInfo {
    boost::uuids::uuid id;
    QString name;
    QString party_category;       // "System" or "Operational"
    QString business_center_code; // FpML code, e.g. "GBLO", "USNY"
 
    bool is_system() const { return party_category == "System"; }
};
 
struct LoginResult {
    bool success = false;
    QString error_message;
    bool password_reset_required = false;
    bool bootstrap_mode = false;
    bool tenant_bootstrap_mode = false;
    bool party_setup_required = false;
    boost::uuids::uuid selected_party_id;
    std::vector<PartyInfo> available_parties;
};
 
struct SignupResult {
    bool success = false;
    QString error_message;
    QString username;
};
 
struct SessionListResult {
    std::vector<iam::domain::session> sessions;
    std::uint32_t total_count = 0;
};
 
struct TradeListResult {
    std::vector<trading::domain::trade> trades;
    std::uint32_t total_count = 0;
};
 
class ORES_QT_API ClientManager : public QObject {
    Q_OBJECT
 
private:
    inline static std::string_view logger_name =
        "ores.qt.client_manager";
 
    [[nodiscard]] static auto& lg() {
        using namespace ores::logging;
        static auto instance = make_logger(logger_name);
        return instance;
    }
 
public:
    // Standard timeout for quick NATS round-trips (queries, lookups).
    static constexpr std::chrono::seconds fast_timeout{30};
    // Extended timeout for slow server-side operations (provisioning, imports).
    static constexpr std::chrono::seconds slow_timeout{120};
 
    explicit ClientManager(std::shared_ptr<eventing::service::event_bus> event_bus,
                           QObject* parent = nullptr);
    ~ClientManager() override;
 
    void setSubjectPrefix(const std::string& prefix) { subject_prefix_ = prefix; }
 
    const std::string& subjectPrefix() const { return subject_prefix_; }
 
    LoginResult connect(const std::string& host, std::uint16_t port);
 
    LoginResult login(const std::string& username, const std::string& password);
 
    LoginResult connectAndLogin(
        const std::string& host,
        std::uint16_t port,
        const std::string& username,
        const std::string& password);
 
    LoginResult testConnection(
        const std::string& host,
        std::uint16_t port,
        const std::string& username,
        const std::string& password);
 
    SignupResult signup(
        const std::string& host,
        std::uint16_t port,
        const std::string& username,
        const std::string& email,
        const std::string& password);
 
    void disconnect();
 
    bool logout();
 
    bool isConnected() const;
 
    [[deprecated("Permission checks are now server-side via RBAC")]]
    bool isAdmin() const { return false; }
 
    bool isLoggedIn() const { return session_.is_logged_in(); }
 
    [[nodiscard]] nats::service::jetstream_admin admin();
 
    std::string currentUsername() const {
        if (!session_.is_logged_in()) return {};
        return session_.auth().username;
    }
 
    std::string currentEmail() const { return current_email_; }
 
    void setCurrentEmail(const std::string& email) { current_email_ = email; }
 
    std::optional<boost::uuids::uuid> accountId() const { return current_account_id_; }
 
    std::string serverAddress() const {
        if (!isConnected()) return "";
        return connected_host_ + ":" + std::to_string(connected_port_);
    }
 
    std::string connectedHost() const {
        if (!isConnected()) return "";
        return connected_host_;
    }
 
    std::uint16_t connectedPort() const {
        if (!isConnected()) return 0;
        return connected_port_;
    }
 
    std::string storedUsername() const { return stored_username_; }
 
    std::string storedPassword() const { return stored_password_; }
 
    boost::uuids::uuid currentPartyId() const { return current_party_id_; }
 
    QString currentPartyName() const { return current_party_name_; }
 
    QString currentPartyCategory() const { return current_party_category_; }
 
    bool isSystemParty() const { return current_party_category_ == "System"; }
 
    bool lastPartySetupRequired() const { return last_party_setup_required_; }
 
    const std::string& httpBaseUrl() const { return http_base_url_; }
 
    bool selectParty(const boost::uuids::uuid& party_id, const QString& party_name);
 
    template <nats_request RequestType>
    auto process_request(RequestType request)
        -> std::expected<typename RequestType::response_type, std::string> {
        using ResponseType = typename RequestType::response_type;
        try {
            const auto json_body = rfl::json::write(request);
            auto msg = session_.request(RequestType::nats_subject, json_body);
            const std::string_view data(
                reinterpret_cast<const char*>(msg.data.data()),
                msg.data.size());
            auto result = rfl::json::read<ResponseType, rfl::AddTagsToVariants>(data);
            if (!result) {
                return std::unexpected(
                    std::string("Failed to deserialize response: ") +
                    result.error().what());
            }
            return std::move(*result);
        } catch (const ores::nats::service::nats_connect_error&) {
            throw;  // Propagate so connect() can map to a user-visible message
        } catch (const std::exception& e) {
            return std::unexpected(std::string(e.what()));
        }
    }
 
    template <nats_request RequestType>
    auto process_authenticated_request(RequestType request,
        std::chrono::milliseconds timeout = std::chrono::seconds(30))
        -> std::expected<typename RequestType::response_type, std::string> {
        using ResponseType = typename RequestType::response_type;
        if (!session_.is_logged_in()) {
            return std::unexpected(std::string("Not logged in"));
        }
        try {
            const auto cid = boost::uuids::to_string(
                boost::uuids::random_generator()());
            const auto json_body = rfl::json::write(request);
            auto scoped = session_
                .with_correlation_id(cid)
                .with_session_id(session_id_);
            auto msg = scoped.authenticated_request(
                RequestType::nats_subject, json_body, timeout);
            const std::string_view data(
                reinterpret_cast<const char*>(msg.data.data()),
                msg.data.size());
            auto result = rfl::json::read<ResponseType, rfl::AddTagsToVariants>(data);
            if (!result) {
                return std::unexpected(
                    std::string("Failed to deserialize response: ") +
                    result.error().what());
            }
            return std::move(*result);
        } catch (const ores::nats::service::nats_connect_error&) {
            throw;  // Propagate so connect() can map to a user-visible message
        } catch (const ores::nats::service::session_expired_error& e) {
            using namespace ores::logging;
            BOOST_LOG_SEV(lg(), warn) << "Session expired: " << e.what();
            QMetaObject::invokeMethod(this, "sessionExpired", Qt::QueuedConnection);
            return std::unexpected(std::string(e.what()));
        } catch (const std::exception& e) {
            return std::unexpected(std::string(e.what()));
        }
    }
 
    std::optional<SessionListResult> listSessions(
        const boost::uuids::uuid& accountId,
        std::uint32_t limit = 100,
        std::uint32_t offset = 0);
 
    std::optional<TradeListResult> listTrades(
        std::optional<boost::uuids::uuid> node_id = std::nullopt,
        std::uint32_t offset = 0,
        std::uint32_t limit = 100);
 
    std::optional<trading::messaging::trade_export_item>
    getTradeDetail(const std::string& trade_id);
 
    std::optional<std::vector<iam::domain::session>> getActiveSessions();
 
    std::optional<std::vector<iam::messaging::session_sample_dto>>
    getSessionSamples(const boost::uuids::uuid& sessionId);
 
    // =========================================================================
    // Connection Status Accessors (stubbed for NATS - not all available)
    // =========================================================================
 
    std::uint64_t bytesSent() const { return 0; }
    std::uint64_t bytesReceived() const { return 0; }
    std::uint64_t lastRttMs() const { return 0; }
 
    std::optional<std::chrono::steady_clock::time_point> disconnectedSince() const {
        return disconnected_since_;
    }
 
    // =========================================================================
    // Session Recording (stubbed - not applicable for NATS)
    // =========================================================================
 
    bool enableRecording(const std::filesystem::path&) { return false; }
    void disableRecording() {}
    bool isRecording() const { return false; }
    std::filesystem::path recordingFilePath() const { return {}; }
    void setRecordingDirectory(const std::filesystem::path& dir) {
        recording_directory_ = dir;
    }
    std::filesystem::path recordingDirectory() const {
        return recording_directory_;
    }
 
    // =========================================================================
    // Event Subscriptions
    // =========================================================================
 
    void subscribeToEvent(const std::string& subject);
    void unsubscribeFromEvent(const std::string& subject);
 
signals:
    void connected();
    void loggedIn();
    void disconnected();
    void reconnecting();
    void reconnected();
    void connectionError(const QString& message);
 
    void sessionExpired();
 
    void notificationReceived(const QString& eventType, const QDateTime& timestamp,
                              const QStringList& entityIds, const QString& tenantId,
                              int payloadType, const QByteArray& payload);
 
    void recordingStarted(const QString& filePath);
    void recordingStopped();
    void streamingStarted();
    void streamingStopped();
 
private:
    // Fraction of token lifetime at which the proactive refresh fires
    static constexpr double refresh_lifetime_ratio = 0.8;
 
    void arm_refresh_timer(int lifetime_s);
 
    void onRefreshTimer();
 
    // NATS session for connection and authentication
    ores::nats::service::nats_client session_;
 
    // Subject prefix applied to all outbound NATS messages
    std::string subject_prefix_;
 
    // Event bus for publishing connection events
    std::shared_ptr<eventing::service::event_bus> event_bus_;
 
    // Connection details
    std::string connected_host_;
    std::uint16_t connected_port_{0};
 
    // Session recording directory
    std::filesystem::path recording_directory_;
 
    // Time point when the connection was lost
    std::optional<std::chrono::steady_clock::time_point> disconnected_since_;
 
    // Stored credentials for display/reconnect
    std::string stored_username_;
    std::string stored_password_;
 
    // Current account info (from login)
    std::optional<boost::uuids::uuid> current_account_id_;
    std::string current_email_;
 
    // Session trace ID generated on login, forwarded on every NATS request
    // as Nats-Session-Id to group all calls from this login session in logs.
    std::string session_id_;
 
    // Currently selected party context
    boost::uuids::uuid current_party_id_;
    QString current_party_name_;
    QString current_party_category_;
    bool last_party_setup_required_ = false;
 
    // HTTP server base URL discovered post-login via NATS.
    std::string http_base_url_;
 
    // Active NATS event subscriptions keyed by subject
    std::unordered_map<std::string, nats::service::subscription> nats_subscriptions_;
 
    // Proactive token refresh timer (fires before token expires)
    QTimer* refresh_timer_ = nullptr;
};
 
}
 
#endif