TwelveWSClient.cpp

#include "TwelveWSClient.hpp"
#include "Logger.hpp"
#include <json/json.h>
#include <chrono>
#include <format>
#if defined(__APPLE__)
#include <mach/mach_time.h>
#include <mach/mach.h>
#include <mach/thread_policy.h>
#include <pthread.h>
#endif
#include <memory>
#include <string_view>
#include <type_traits>
#include <utility>
#include <vector>

namespace {
std::string stringifyJson(const Json::Value& value) {
    Json::StreamWriterBuilder builder;
    builder["indentation"] = "";
    return Json::writeString(builder, value);
}

}

TwelveWSClient::TwelveWSClient(const std::string& apiKey)
    : mApiKey(apiKey) {
    mJsonReaderBuilder["collectComments"] = false;
    mJsonReaderBuilder["allowDroppedNullPlaceholders"] = false;
    mJsonReader = std::unique_ptr<Json::CharReader>(mJsonReaderBuilder.newCharReader());
}

TwelveWSClient::~TwelveWSClient() {
    stop();
}

void TwelveWSClient::addSymbol(std::string_view symbol) {
    std::scoped_lock lock(mMtx);
    mSymbolQueue.emplace(symbol);
}

void TwelveWSClient::start() {
    if (mRunning.exchange(true)) {
        Logger::instance().info("[CLIENT] WebSocket client already running");
        return;
    }
    Logger::instance().info("[CLIENT] Starting WebSocket client");
    mConnect();
    mHeartbeatThread = std::jthread([this](std::stop_token stopToken) {
        mHeartbeatLoop(stopToken);
    });
}

void TwelveWSClient::stop() {
    if (!mRunning.exchange(false)) {
        return;
    }

    Logger::instance().info("[CLIENT] Stopping WebSocket client");
    // Unsubscribe from all active symbols before closing
    std::vector<std::string> toUnsub;
    {
        std::scoped_lock lock(mMtx);
        for (const auto& sym : mActiveSymbols) {
            toUnsub.push_back(sym);
        }
        mActiveSymbols.clear();
    }
    if (!toUnsub.empty()) {
        std::string symbolList;
        for (size_t i = 0; i < toUnsub.size(); ++i) {
            symbolList += toUnsub[i];
            if (i + 1 < toUnsub.size()) symbolList += ",";
        }
        Json::Value unsub;
        unsub["action"] = "unsubscribe";
        unsub["params"]["symbols"] = symbolList;
        std::string payload = stringifyJson(unsub);
        Logger::instance().info(std::format("[UNSUB] Sending payload: {}", payload));
        mWs.send(payload);
    }
    if (mHeartbeatThread.joinable()) {
        mHeartbeatThread.request_stop();
        mHeartbeatThread.join();
    }
    mWs.close();
}

void TwelveWSClient::mConnect() {
    const std::string baseUrl = "wss://ws.twelvedata.com/v1/quotes/price";
    std::string url;
    url.reserve(baseUrl.size() + 9 + mApiKey.size());
    url.append(baseUrl);
    url.append("?apikey=");
    url.append(mApiKey);
    Logger::instance().info("[CLIENT] Connecting to " + baseUrl + "?apikey=***");
    mWs.setUrl(url);

    // Configure message handler
    mWs.setOnMessageCallback([this](const ix::WebSocketMessagePtr& msg) {
        if (msg->type == ix::WebSocketMessageType::Open) {
            Logger::instance().info("[WS] Connected");
            mSubscribeNext();
        }
        else if (msg->type == ix::WebSocketMessageType::Message) {
            mHandleMessage(msg->str);
        }
        else if (msg->type == ix::WebSocketMessageType::Close) {
            Logger::instance().info(std::format("[WS] Closed code={} reason={} remote={}",
                                               msg->closeInfo.code,
                                               msg->closeInfo.reason,
                                               msg->closeInfo.remote));
        }
        else if (msg->type == ix::WebSocketMessageType::Error) {
            Logger::instance().error(std::format(
                "[WS] Error message: {} http_status={} retries={} wait_time={} decompressionError={}",
                msg->errorInfo.reason,
                msg->errorInfo.http_status,
                msg->errorInfo.retries,
                msg->errorInfo.wait_time,
                msg->errorInfo.decompressionError));
        }
        else if (msg->type == ix::WebSocketMessageType::Ping) {
            Logger::instance().info("[WS] Ping received");
        }
        else if (msg->type == ix::WebSocketMessageType::Pong) {
            Logger::instance().info("[WS] Pong received");
        }
        else {
            using Enum = ix::WebSocketMessageType;
            const auto typeValue = static_cast<std::underlying_type_t<Enum>>(msg->type);
            Logger::instance().info(std::format("[WS] Other message type: {}", typeValue));
        }
    });

    Logger::instance().info("[CLIENT] Starting ix::WebSocket event loop");
    mWs.start();
}

void TwelveWSClient::mSubscribeNext() {
    struct PendingSubscription {
        std::string symbol;
        std::string payload;
    };

    std::vector<std::string> batch;
    {
        std::unique_lock lock(mMtx);
        while (mActiveSymbols.size() < mMaxConcurrent && !mSymbolQueue.empty()) {
            std::string sym = std::move(mSymbolQueue.front());
            mSymbolQueue.pop();
            mActiveSymbols.insert(sym);
            batch.push_back(sym);
        }
    }
    if (!batch.empty()) {
        std::string symbolList;
        for (size_t i = 0; i < batch.size(); ++i) {
            symbolList += batch[i];
            if (i + 1 < batch.size()) symbolList += ",";
        }
        Json::Value sub;
        sub["action"] = "subscribe";
        sub["params"]["symbols"] = symbolList;
        std::string payload = stringifyJson(sub);
        Logger::instance().info(std::format("[SUB] Sending payload: {}", payload));
        mWs.send(payload);
        Logger::instance().info(std::format("[SUB] {}", symbolList));
    }
}

void TwelveWSClient::mHandleMessage(const std::string& msg) {
    // Only log raw message for non-price events (for debugging)
    Json::Value data;
    std::string errs;
    {
    std::scoped_lock readerLock(mJsonReaderMutex);
    if (!mJsonReader) {
        mJsonReader = std::unique_ptr<Json::CharReader>(mJsonReaderBuilder.newCharReader());
    }
    if (!mJsonReader->parse(msg.data(), msg.data() + msg.size(), &data, &errs)) {
        Logger::instance().error("[ERROR] JSON parse failed: " + errs + " | msg=" + msg);
        return;
    }
    }
    // Remove all '[WS] Raw message:' logs for price events and legacy code
    // Update last message time
    mLastMessageTime = std::chrono::steady_clock::now();
    // ...existing code...
    // Update last message time
    mLastMessageTime = std::chrono::steady_clock::now();
    try {
        Json::Value data;
        std::string errs;
        {
            std::scoped_lock readerLock(mJsonReaderMutex);
            if (!mJsonReader) {
                mJsonReader = std::unique_ptr<Json::CharReader>(mJsonReaderBuilder.newCharReader());
            }

            if (!mJsonReader->parse(msg.data(), msg.data() + msg.size(), &data, &errs)) {
                Logger::instance().error("[ERROR] JSON parse failed: " + errs + " | msg=" + msg);
                return;
            }
        }

        if (data.isMember("event")) {
            const std::string event = data["event"].asString();
            if (event == "connected") {
                Logger::instance().info("[WS] Server acknowledged connection");
            }
            else if (event == "subscribe-status") {
                const Json::Value& success = data["success"];
                if (!success.isNull()) {
                    Logger::instance().info(std::format("[SUB] Success payload: {}", stringifyJson(success)));
                }

                const Json::Value& fails = data["fails"];
                if (const bool hasFails = !fails.isNull() && fails.isArray() && !fails.empty(); hasFails) {
                    Logger::instance().error(std::format("[SUB] Subscription failures: {}", stringifyJson(fails)));
                    {
                        std::scoped_lock lock(mMtx);
                        for (const auto& entry : fails) {
                            if (entry.isMember("symbol")) {
                                mActiveSymbols.erase(entry["symbol"].asString());
                            }
                        }
                    }
                    // Attempt to fill the subscription window with remaining queued symbols after releasing the lock.
                    mSubscribeNext();
                }
            }
            else if (event == "error") {
                Logger::instance().error("[WS] Error event payload: " + msg);
            }
        }

        // Handle real-time price updates
        if (data.isMember("symbol")) {
            std::string symbol = data["symbol"].asString();
            double price = 0.0;
            const Json::Value& priceVal = data["price"];
            if (priceVal.isDouble() || priceVal.isIntegral()) {
                price = priceVal.asDouble();
            }
            else if (priceVal.isString()) {
                price = std::stod(priceVal.asString());
            }
            else {
                Logger::instance().error("[ERROR] Unexpected price format | msg=" + msg);
                return;
            }
            std::string currency;
            if (data.isMember("currency")) {
                currency = data["currency"].asString();
            } else if (data.isMember("currency_quote")) {
                currency = data["currency_quote"].asString();
            }
            std::ostringstream priceStream;
            priceStream << std::fixed << std::setprecision(4) << price;
            std::string priceStr = priceStream.str();
            if (!currency.empty()) {
                    Logger::instance().update(std::format("{}: {} {}", symbol, priceStr, currency));
            } else {
                    Logger::instance().update(std::format("{}: {}", symbol, priceStr));
            }

            // Stay subscribed: do not remove symbol, do not unsubscribe, do not call mSubscribeNext().
            // (No action here)
        }
    } 
    catch (const std::exception& e) {
        Logger::instance().error(std::string{"[ERROR] "} + e.what() + " | msg=" + msg);
    }
}

void TwelveWSClient::mHeartbeatLoop(std::stop_token stopToken) {
    using clock = std::chrono::steady_clock;
    auto nextTick = clock::now() + mHeartbeatInterval;
    while (!stopToken.stop_requested() && mRunning.load()) {
        std::this_thread::sleep_until(nextTick);
        if (stopToken.stop_requested() || !mRunning.load()) {
            break;
        }
        // Only send heartbeat if no message received in last 10 seconds
        auto now = clock::now();
        auto lastMsg = mLastMessageTime.load();
        auto sinceLastMsg = std::chrono::duration_cast<std::chrono::seconds>(now - lastMsg);
        if (mWs.getReadyState() == ix::ReadyState::Open && sinceLastMsg >= std::chrono::seconds(10)) {
            Logger::instance().info("[HB] Sending heartbeat");
            mWs.send(mHeartbeatPayload);
        }
        nextTick += mHeartbeatInterval;
    }
}