From d43979b63826fb4e34c3164428862e093a3a72f6 Mon Sep 17 00:00:00 2001 From: Aster Seker Date: Fri, 17 Jul 2026 12:48:57 +0300 Subject: [PATCH] feat(bridges): add metatrader file bridge --- .github/workflows/ubuntu-smoke.yml | 8 +- .github/workflows/windows-smoke.yml | 5 + CMakeLists.txt | 46 + .../metatrader_file_bridge_smoke.config.json | 19 + examples/metatrader_file_bridge_smoke.cpp | 302 ++++ .../file-transport-and-adapters.md | 15 +- .../file-transport-and-adapters.ru.md | 15 +- .../optionx_cpp/bridges/metatrader_file.hpp | 10 +- .../metatrader_file/MetaTraderFileBridge.hpp | 1442 +++++++++++++++++ .../MetaTraderFileBridgeConfig.hpp | 71 + .../detail/MetaTraderFileBridgeUtils.hpp | 446 +++++ .../detail/MetaTraderFileIdempotencyStore.hpp | 235 +++ .../detail/MetaTraderFileOperationKey.hpp | 40 + .../{ => detail}/MetaTraderFilePathUtils.hpp | 6 +- .../detail/MetaTraderFileProtocol.hpp | 1 + tests/bridge_umbrella_include_test.cpp | 2 + tests/metatrader_file_bridge_test.cpp | 1250 ++++++++++++++ 17 files changed, 3896 insertions(+), 17 deletions(-) create mode 100644 examples/metatrader_file_bridge_smoke.config.json create mode 100644 examples/metatrader_file_bridge_smoke.cpp create mode 100644 include/optionx_cpp/bridges/metatrader_file/MetaTraderFileBridge.hpp create mode 100644 include/optionx_cpp/bridges/metatrader_file/detail/MetaTraderFileBridgeUtils.hpp create mode 100644 include/optionx_cpp/bridges/metatrader_file/detail/MetaTraderFileIdempotencyStore.hpp create mode 100644 include/optionx_cpp/bridges/metatrader_file/detail/MetaTraderFileOperationKey.hpp rename include/optionx_cpp/bridges/metatrader_file/{ => detail}/MetaTraderFilePathUtils.hpp (93%) diff --git a/.github/workflows/ubuntu-smoke.yml b/.github/workflows/ubuntu-smoke.yml index 4c55f5e..23b5c40 100644 --- a/.github/workflows/ubuntu-smoke.yml +++ b/.github/workflows/ubuntu-smoke.yml @@ -22,7 +22,7 @@ jobs: sudo apt-get install -y libssl-dev libcurl4-openssl-dev - name: Configure - run: cmake -S . -B build-linux -DOPTIONX_BUILD_DEPS=ON -DOPTIONX_BUILD_TESTS=ON + run: cmake -S . -B build-linux -DOPTIONX_BUILD_DEPS=ON -DOPTIONX_BUILD_TESTS=ON -DOPTIONX_BUILD_EXAMPLES=ON - name: Build trade_record_db_test run: cmake --build build-linux --target trade_record_db_test -j @@ -54,6 +54,12 @@ jobs: - name: Run metatrader_file_bridge_test run: ./build-linux/metatrader_file_bridge_test --gtest_brief=1 + - name: Build metatrader_file_bridge_smoke + run: cmake --build build-linux --target metatrader_file_bridge_smoke -j + + - name: Run metatrader_file_bridge_smoke + run: ./build-linux/metatrader_file_bridge_smoke --self-test + - name: Build market_data_subscription_contract_test run: cmake --build build-linux --target market_data_subscription_contract_test -j diff --git a/.github/workflows/windows-smoke.yml b/.github/workflows/windows-smoke.yml index c050297..9b375bf 100644 --- a/.github/workflows/windows-smoke.yml +++ b/.github/workflows/windows-smoke.yml @@ -21,6 +21,7 @@ jobs: cmake -S . -B build-windows -DOPTIONX_BUILD_DEPS=ON -DOPTIONX_BUILD_TESTS=ON + -DOPTIONX_BUILD_EXAMPLES=ON -DOPTIONX_LIGHTWEIGHT_BRIDGE_SMOKE_TESTS=ON - name: Build MetaTrader path discovery test @@ -33,6 +34,9 @@ jobs: metatrader_file_bridge_test bridge_umbrella_include_test + - name: Build MetaTrader file smoke example + run: cmake --build build-windows --config Debug --target metatrader_file_bridge_smoke + - name: Run MetaTrader file tests shell: pwsh run: | @@ -41,3 +45,4 @@ jobs: .\build-windows\Debug\metatrader_file_config_include_test.exe --gtest_brief=1 .\build-windows\Debug\metatrader_file_bridge_test.exe --gtest_brief=1 .\build-windows\Debug\bridge_umbrella_include_test.exe --gtest_brief=1 + .\build-windows\Debug\metatrader_file_bridge_smoke.exe --self-test diff --git a/CMakeLists.txt b/CMakeLists.txt index df8dd79..517387a 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -439,6 +439,52 @@ if(OPTIONX_BUILD_EXAMPLES) -P "${CMAKE_CURRENT_SOURCE_DIR}/cmake/copy_runtime_dlls.cmake" ) endif() + + add_executable(metatrader_file_bridge_smoke examples/metatrader_file_bridge_smoke.cpp) + target_compile_features(metatrader_file_bridge_smoke PRIVATE cxx_std_17) + + target_include_directories(metatrader_file_bridge_smoke PRIVATE + ${EXAMPLE_INCLUDE_DIRS} + ${EXAMPLE_DEPS_INCLUDE_DIRS} + ) + + target_link_directories(metatrader_file_bridge_smoke PRIVATE ${EXAMPLE_LIBRARY_DIRS}) + target_compile_definitions( + metatrader_file_bridge_smoke PRIVATE + ${EXAMPLE_DEFINES} + LOGIT_BASE_PATH="${LOGIT_BASE_PATH_FWD}" + ) + if(MINGW) + target_compile_options(metatrader_file_bridge_smoke PRIVATE -Wa,-mbig-obj) + endif() + if(OPTIONX_LIGHTWEIGHT_BRIDGE_SMOKE_TESTS) + target_link_libraries(metatrader_file_bridge_smoke PRIVATE ${EXAMPLE_LIBS}) + if(WIN32) + target_link_libraries(metatrader_file_bridge_smoke PRIVATE ${OPTIONX_WINDOWS_SYSTEM_LIBS}) + endif() + else() + target_link_libraries(metatrader_file_bridge_smoke PRIVATE ${EXAMPLE_LIBS} optionx_cpp) + endif() + + if(OPTIONX_BUILD_DEPS AND NOT OPTIONX_LIGHTWEIGHT_BRIDGE_SMOKE_TESTS) + add_dependencies(metatrader_file_bridge_smoke mdbx-static AES) + endif() + + foreach(dll ${EXAMPLE_DLL_FILES}) + add_custom_command(TARGET metatrader_file_bridge_smoke POST_BUILD + COMMAND ${CMAKE_COMMAND} -E copy_if_different + "${dll}" "$" + ) + endforeach() + + if(WIN32) + add_custom_command(TARGET metatrader_file_bridge_smoke POST_BUILD + COMMAND ${CMAKE_COMMAND} + -DOPTIONX_RUNTIME_DLL_DIR="${EXAMPLE_BUILD_LIBS_DIR}/bin" + -DOPTIONX_RUNTIME_TARGET_DIR="$" + -P "${CMAKE_CURRENT_SOURCE_DIR}/cmake/copy_runtime_dlls.cmake" + ) + endif() endif() if(OPTIONX_BUILD_TESTS) diff --git a/examples/metatrader_file_bridge_smoke.config.json b/examples/metatrader_file_bridge_smoke.config.json new file mode 100644 index 0000000..3814bcc --- /dev/null +++ b/examples/metatrader_file_bridge_smoke.config.json @@ -0,0 +1,19 @@ +{ + // Omit common_files_root to use the default MetaQuotes Common\Files folder. + // For portable terminals, set it explicitly, for example: + // "common_files_root": "C:/MetaTrader/Terminal/Common/Files", + + "namespace_subdir": "OptionX/Bridge/v1", + "bridge_id": 1, + "client_id": "smoke", + "client_secret": "", + + "poll_interval_ms": 250, + "max_line_bytes": 65536, + "max_scanned_records_per_poll": 256, + "max_returned_records_per_poll": 64, + "idempotency_retention_ms": 86400000, + + "enable_events": true, + "enable_state_snapshot": true +} diff --git a/examples/metatrader_file_bridge_smoke.cpp b/examples/metatrader_file_bridge_smoke.cpp new file mode 100644 index 0000000..e998325 --- /dev/null +++ b/examples/metatrader_file_bridge_smoke.cpp @@ -0,0 +1,302 @@ +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#ifdef _WIN32 +#ifndef WIN32_LEAN_AND_MEAN +#define WIN32_LEAN_AND_MEAN +#endif +#include +#endif + +namespace { + +using optionx::bridges::metatrader_file::MetaTraderFileBridge; +using optionx::bridges::metatrader_file::MetaTraderFileBridgeConfig; + +std::atomic_bool g_stop_requested{false}; +std::atomic_int g_interrupt_count{0}; + +bool has_arg(int argc, char** argv, const std::string& value) { + for (int i = 1; i < argc; ++i) { + if (argv[i] == value) { + return true; + } + } + return false; +} + +std::string option_value(int argc, char** argv, const std::string& name) { + for (int i = 1; i + 1 < argc; ++i) { + if (argv[i] == name) { + return argv[i + 1]; + } + } + return {}; +} + +std::filesystem::path smoke_root() { + return std::filesystem::temp_directory_path() / "optionx-metatrader-file-smoke"; +} + +MetaTraderFileBridgeConfig default_config(const bool self_test) { + MetaTraderFileBridgeConfig config; + config.bridge_id = 1; + config.client_id = "smoke"; + config.poll_interval_ms = 100; + config.max_scanned_records_per_poll = 128; + config.max_returned_records_per_poll = 32; + if (self_test || config.common_files_root.empty()) { + config.common_files_root = smoke_root().u8string(); + } + return config; +} + +bool load_config( + const std::string& path, + MetaTraderFileBridgeConfig& config) { + if (path.empty()) { + return true; + } + + std::ifstream input(path); + if (!input) { + std::cerr << "Could not open config: " << path << '\n'; + return false; + } + + try { + std::ostringstream buffer; + buffer << input.rdbuf(); + const auto json = optionx::utils::parse_json_with_comments(buffer.str()); + config.from_json(json); + } catch (const std::exception& ex) { + std::cerr << "Could not parse config: " << ex.what() << '\n'; + return false; + } + return true; +} + +nlohmann::json make_self_test_command() { + const auto operation_key = + optionx::bridges::metatrader_file::make_compact_operation_key("smoke"); + return nlohmann::json{ + {"file_seq", 1}, + {"jsonrpc", "2.0"}, + {"id", operation_key}, + {"method", "signal.submit"}, + {"params", { + {"context", { + {"idempotency_key", operation_key}, + {"valid_until_ms", + optionx::bridges::metatrader_file::detail::unix_time_ms() + 60000} + }}, + {"identity", { + {"signal_name", "metatrader_file_smoke"}, + {"unique_hash", operation_key} + }}, + {"signal", { + {"symbol", "EURUSD"}, + {"order_type", "BUY"}, + {"option_type", "SPRINT"}, + {"amount", { + {"value", "1.00"}, + {"currency", "USD"} + }}, + {"expiry", { + {"kind", "duration"}, + {"duration_ms", 60000} + }} + }} + }} + }; +} + +bool append_command_line( + const std::filesystem::path& commands_log, + const nlohmann::json& command) { + try { + std::filesystem::create_directories(commands_log.parent_path()); + std::ofstream out(commands_log, std::ios::binary | std::ios::app); + if (!out) { + std::cerr << "Could not open commands log: " + << commands_log.u8string() << '\n'; + return false; + } + out << command.dump(-1) << '\n'; + } catch (const std::exception& ex) { + std::cerr << "Could not append command: " << ex.what() << '\n'; + return false; + } + return true; +} + +std::string read_text_file(const std::filesystem::path& path) { + std::ifstream input(path, std::ios::binary); + if (!input) { + return {}; + } + std::ostringstream buffer; + buffer << input.rdbuf(); + return buffer.str(); +} + +void request_stop_from_interrupt() { + const auto count = g_interrupt_count.fetch_add(1) + 1; + if (count == 1) { + g_stop_requested.store(true); + return; + } + std::_Exit(130); +} + +#ifdef _WIN32 +BOOL WINAPI console_ctrl_handler(DWORD event_type) { + switch (event_type) { + case CTRL_C_EVENT: + case CTRL_BREAK_EVENT: + case CTRL_CLOSE_EVENT: + case CTRL_LOGOFF_EVENT: + case CTRL_SHUTDOWN_EVENT: + request_stop_from_interrupt(); + return TRUE; + default: + return FALSE; + } +} +#else +void signal_handler(int) { + request_stop_from_interrupt(); +} +#endif + +void install_stop_handlers() { + g_stop_requested.store(false); + g_interrupt_count.store(0); +#ifdef _WIN32 + SetConsoleCtrlHandler(console_ctrl_handler, TRUE); +#else + std::signal(SIGINT, signal_handler); + std::signal(SIGTERM, signal_handler); +#endif +} + +} // namespace + +int main(int argc, char** argv) { + const bool self_test = has_arg(argc, argv, "--self-test"); + install_stop_handlers(); + + auto config = default_config(self_test); + if (!load_config(option_value(argc, argv, "--config"), config)) { + return 2; + } + + const auto validation = config.validate(); + if (!validation.first) { + std::cerr << "Invalid config: " << validation.second << '\n'; + return 2; + } + + if (self_test) { + std::error_code ec; + std::filesystem::remove_all(config.client_root(), ec); + } + + MetaTraderFileBridge bridge; + if (!bridge.configure(std::make_unique(config))) { + std::cerr << "Bridge configuration failed\n"; + return 2; + } + + std::atomic next_signal_id{1}; + std::atomic received_signals{0}; + + bridge.on_signal_id() = [&next_signal_id]() { + return next_signal_id.fetch_add(1); + }; + bridge.on_status_update() = [](const optionx::BridgeStatusUpdate& update) { + std::cout << "status=" << optionx::to_str(update.status); + if (!update.connection_id.empty()) { + std::cout << " connection=" << update.connection_id; + } + if (!update.message.empty()) { + std::cout << " message=" << update.message; + } + std::cout << '\n'; + }; + bridge.on_trade_signal() = [&received_signals](std::unique_ptr signal) { + ++received_signals; + nlohmann::json json = *signal; + std::cout << "signal:\n" << json.dump(2) << '\n'; + }; + bridge.on_signal_report() = [](const optionx::BridgeSignalReport& report) { + nlohmann::json json = report; + std::cout << "signal_report:\n" << json.dump(2) << '\n'; + }; + + const auto root = config.client_root(); + const auto commands_log = root / "commands.ndjson"; + const auto events_log = root / "events.ndjson"; + + std::cout << "MetaTrader file bridge root: " << root.u8string() << '\n'; + std::cout << "MQL writes commands to: " << commands_log.u8string() << '\n'; + std::cout << "OptionX writes events to: " << events_log.u8string() << '\n'; + + if (self_test) { + if (!append_command_line(commands_log, make_self_test_command())) { + return 3; + } + + try { + bridge.process(); + } catch (const std::exception& ex) { + std::cerr << "Bridge process failed: " << ex.what() << '\n'; + return 4; + } + + const auto events = read_text_file(events_log); + if (!events.empty()) { + std::cout << "events.ndjson:\n" << events; + } + return received_signals.load() > 0 && events.find("\"status\":\"accepted\"") != std::string::npos + ? 0 + : 5; + } + + bridge.run(); + std::cout << "Press Enter or Ctrl+C to stop...\n"; + + std::atomic_bool input_done{false}; + std::thread input_thread([&input_done]() { + std::string line; + std::getline(std::cin, line); + input_done.store(true); + g_stop_requested.store(true); + }); + + while (!g_stop_requested.load()) { + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + } + + if (input_thread.joinable()) { + if (input_done.load()) { + input_thread.join(); + } else { + input_thread.detach(); + } + } + bridge.shutdown(); + return 0; +} diff --git a/guides/bridge-protocol-v1/file-transport-and-adapters.md b/guides/bridge-protocol-v1/file-transport-and-adapters.md index e89312a..e29680d 100644 --- a/guides/bridge-protocol-v1/file-transport-and-adapters.md +++ b/guides/bridge-protocol-v1/file-transport-and-adapters.md @@ -279,11 +279,14 @@ and the number of returned new records per poll. The scan limit counts accepted, already-seen and malformed complete lines so malformed input cannot make one poll accumulate unbounded diagnostics. -Trade-affecting commands still require `context.idempotency_key`. -`context.valid_until_ms` is strongly recommended because file polling can -introduce delay. If a command with the same JSON-RPC `id` or idempotency key is -seen again, the bridge must return or re-emit the original/current operation -result instead of creating a second trade. +Trade-affecting commands still require `context.idempotency_key`. Concrete +polling bridges should also require `context.valid_until_ms` because file +polling can introduce delay and append logs may be retained longer than the +dedupe window. If a command with the same JSON-RPC `id` or idempotency key is +seen again within the configured idempotency retention window, the bridge must +return or re-emit the original/current operation result instead of creating a +second trade. After that retention window, duplicate suppression is no longer +guaranteed; stale retained commands should be rejected by `valid_until_ms`. This draft intentionally does not define log rotation. Production implementations may add owner-side compaction later, but the baseline profile is @@ -291,7 +294,7 @@ append, checkpoint and clear. Deferred implementation work: -- A concrete polling bridge class built on this protocol helper layer. +- Higher-level MQL helpers for writing and compacting these logs. - A runtime writer object or owner queue that serializes append, repair and owner-side clear operations per log file. - Optional `log_generation`/file identity support if persisted byte-offset diff --git a/guides/bridge-protocol-v1/file-transport-and-adapters.ru.md b/guides/bridge-protocol-v1/file-transport-and-adapters.ru.md index 8b3bf7a..6ceaa32 100644 --- a/guides/bridge-protocol-v1/file-transport-and-adapters.ru.md +++ b/guides/bridge-protocol-v1/file-transport-and-adapters.ru.md @@ -256,18 +256,21 @@ records per poll. Scan limit counts accepted, already-seen and malformed complete lines so malformed input cannot make one poll accumulate unbounded diagnostics. -Trade-affecting commands still require `context.idempotency_key`. -`context.valid_until_ms` is strongly recommended because file polling can add -latency. Repeated command with the same JSON-RPC `id` or idempotency key must -return or re-emit the original/current operation result, not create a second -trade. +Trade-affecting commands still require `context.idempotency_key`. Concrete +polling bridges should also require `context.valid_until_ms`, because file +polling can add latency and append logs may be retained longer than dedupe +window. Repeated command with the same JSON-RPC `id` or idempotency key within +configured idempotency retention window must return or re-emit the +original/current operation result, not create a second trade. After that +retention window duplicate suppression is no longer guaranteed; stale retained +commands should be rejected by `valid_until_ms`. This draft intentionally does not define log rotation. Baseline profile is append, checkpoint and clear. Deferred implementation work: -- Concrete polling bridge class built on this protocol helper layer. +- Higher-level MQL helpers for writing and compacting these logs. - Runtime writer object or owner queue that serializes append, repair and owner-side clear operations per log file. - Optional `log_generation`/file identity support if persisted byte-offset diff --git a/include/optionx_cpp/bridges/metatrader_file.hpp b/include/optionx_cpp/bridges/metatrader_file.hpp index c24f58b..dfefb24 100644 --- a/include/optionx_cpp/bridges/metatrader_file.hpp +++ b/include/optionx_cpp/bridges/metatrader_file.hpp @@ -19,7 +19,9 @@ #include #include #include +#include #include +#include #include #include #include @@ -29,7 +31,9 @@ #include #include +#include #include +#include #if defined(_WIN32) #ifndef NOMINMAX @@ -39,8 +43,12 @@ #endif #include "BaseBridge.hpp" -#include "metatrader_file/MetaTraderFilePathUtils.hpp" +#include "metatrader_file/detail/MetaTraderFilePathUtils.hpp" +#include "metatrader_file/detail/MetaTraderFileOperationKey.hpp" #include "metatrader_file/MetaTraderFileBridgeConfig.hpp" #include "metatrader_file/detail/MetaTraderFileProtocol.hpp" +#include "metatrader_file/detail/MetaTraderFileIdempotencyStore.hpp" +#include "metatrader_file/detail/MetaTraderFileBridgeUtils.hpp" +#include "metatrader_file/MetaTraderFileBridge.hpp" #endif // OPTIONX_HEADER_BRIDGES_METATRADER_FILE_HPP_INCLUDED diff --git a/include/optionx_cpp/bridges/metatrader_file/MetaTraderFileBridge.hpp b/include/optionx_cpp/bridges/metatrader_file/MetaTraderFileBridge.hpp new file mode 100644 index 0000000..f56be17 --- /dev/null +++ b/include/optionx_cpp/bridges/metatrader_file/MetaTraderFileBridge.hpp @@ -0,0 +1,1442 @@ +#pragma once +#ifndef OPTIONX_HEADER_BRIDGES_METATRADER_FILE_META_TRADER_FILE_BRIDGE_HPP_INCLUDED +#define OPTIONX_HEADER_BRIDGES_METATRADER_FILE_META_TRADER_FILE_BRIDGE_HPP_INCLUDED + +/// \file MetaTraderFileBridge.hpp +/// \brief Defines the MetaTrader Common\Files NDJSON bridge. + +namespace optionx::bridges::metatrader_file { + + /// \class MetaTraderFileBridge + /// \brief Polls MetaTrader file-transport commands and publishes bridge events. + /// + /// The bridge owns `events.ndjson` and the command reader checkpoint. The + /// MQL side owns `commands.ndjson`. Incoming `signal.submit` and + /// `trade.open` records are converted to `TradeSignal` snapshots and + /// delivered through `on_trade_signal()`. + class MetaTraderFileBridge final : public BaseBridge { + private: + struct RuntimeState { + std::mutex mutex; + std::shared_ptr account_info; + bridge_status_callback_t status_callback; + BaseBridge::trade_signal_callback_t trade_signal_callback; + BaseBridge::signal_report_callback_t signal_report_callback; + BaseBridge::signal_id_allocator_t signal_id_allocator; + }; + + struct PendingSignalDispatch { + detail::NdjsonRecord record; + nlohmann::json id; + nlohmann::json params; + std::unique_ptr signal; + std::string idempotency_key; + std::string idempotency_storage_key; + std::string request_storage_key; + std::string payload_fingerprint; + }; + + public: + /// \brief Constructs a bridge with empty runtime state. + MetaTraderFileBridge() + : m_state(std::make_shared()) {} + + /// \brief Stops the polling task before destruction. + ~MetaTraderFileBridge() override { + shutdown(); + } + + /// \brief Configures the bridge with MetaTrader file settings. + /// \param config Configuration object. Must be `MetaTraderFileBridgeConfig`. + /// \return `true` when configuration is valid and accepted. + bool configure(std::unique_ptr config) override { + if (!config) return false; + + const auto* typed = + dynamic_cast(config.get()); + if (!typed) { + config->dispatch_callbacks(false, "Invalid MetaTrader file bridge config type."); + return false; + } + + auto next_config = std::make_shared(*typed); + const auto validation = next_config->validate(); + config->dispatch_callbacks(validation.first, validation.second); + if (!validation.first) { + return false; + } + + std::unique_lock config_lock(m_config_mutex, std::defer_lock); + std::unique_lock io_lock(m_io_mutex, std::defer_lock); + std::lock(config_lock, io_lock); + // Runtime paths, offsets and durable sequence state are derived + // once. Reconfiguring after startup would mix old filesystem state + // with a new layout, so it is rejected instead of partially reset. + if (m_runtime_initialized || m_running || m_processing) { + io_lock.unlock(); + config_lock.unlock(); + config->dispatch_callbacks( + false, + "MetaTrader file bridge cannot be reconfigured after runtime start."); + return false; + } + m_config = std::move(next_config); + return true; + } + + /// \brief Returns the status update callback slot. + bridge_status_callback_t& on_status_update() override { + return m_state->status_callback; + } + + /// \brief Returns the trade signal callback slot. + trade_signal_callback_t& on_trade_signal() override { + return m_state->trade_signal_callback; + } + + /// \brief Returns the signal diagnostic report callback slot. + signal_report_callback_t& on_signal_report() override { + return m_state->signal_report_callback; + } + + /// \brief Returns the signal ID allocator slot. + signal_id_allocator_t& on_signal_id() override { + return m_state->signal_id_allocator; + } + + /// \brief Updates account snapshot state and emits balance/state files. + /// \param info Account update received from the trading platform. + void update_account_info(const AccountInfoUpdate& info) override { + if (!info.account_info) return; + + { + std::lock_guard lock(m_state->mutex); + m_state->account_info = info.account_info; + } + + try { + auto config = get_config_or_throw(); + std::lock_guard io_lock(m_io_mutex); + ensure_runtime_started_locked(*config); + if (config->enable_events && detail::should_emit_balance_update(info.status)) { + append_event_locked( + detail::make_balance_updated_notification( + next_event_id_locked("balance"), + detail::source_uri(*config), + m_event_stream_id, + next_event_stream_seq_locked(), + detail::unix_time_ms(), + detail::unix_time_ms(), + detail::account_id_string(*info.account_info), + detail::safe_account_balance(*info.account_info), + detail::safe_account_currency(*info.account_info))); + } + if (config->enable_state_snapshot) { + write_state_snapshot_locked(*config); + } + } catch (const std::exception& ex) { + notify_status(BridgeStatus::CONNECTION_ERROR, {}, ex.what()); + } + } + + /// \brief Emits a trade result update to `events.ndjson`. + /// \param request Original trade request. + /// \param result Current trade result snapshot. + void update_trade_result( + const TradeRequest& request, + const TradeResult& result) override { + try { + auto config = get_config_or_throw(); + std::lock_guard io_lock(m_io_mutex); + ensure_runtime_started_locked(*config); + if (!config->enable_events) { + return; + } + const auto now = detail::unix_time_ms(); + append_event_locked( + detail::make_trade_updated_notification( + next_event_id_locked("trade"), + detail::source_uri(*config), + m_event_stream_id, + next_event_stream_seq_locked(), + result.close_date > 0 ? result.close_date : now, + now, + request, + result)); + } catch (const std::exception& ex) { + notify_status(BridgeStatus::CONNECTION_ERROR, {}, ex.what()); + } + } + + /// \brief Starts periodic polling of `commands.ndjson`. + void run() override { + auto config = get_config_or_throw(); + if (!get_signal_id_allocator()) { + notify_status( + BridgeStatus::SERVER_START_FAILED, + {}, + "MetaTrader file bridge requires a signal ID allocator."); + return; + } + if (!get_trade_signal_callback()) { + notify_status( + BridgeStatus::SERVER_START_FAILED, + {}, + "MetaTrader file bridge requires a trade signal callback."); + return; + } + + { + std::lock_guard io_lock(m_io_mutex); + if (m_running) { + return; + } + try { + ensure_runtime_started_locked(*config); + } catch (const std::exception& ex) { + notify_status(BridgeStatus::SERVER_START_FAILED, {}, ex.what()); + return; + } + m_running = true; + } + + if (!m_task_manager.add_periodic_task( + "metatrader-file-bridge-poll", + config->poll_interval_ms, + [this](std::shared_ptr task) { + if (task->is_shutdown()) return; + try { + process(); + } catch (const std::exception& ex) { + notify_status(BridgeStatus::CONNECTION_ERROR, {}, ex.what()); + } catch (...) { + notify_status( + BridgeStatus::CONNECTION_ERROR, + {}, + "MetaTrader file bridge polling failed with an unknown exception."); + } + })) { + { + std::lock_guard io_lock(m_io_mutex); + m_running = false; + } + notify_status( + BridgeStatus::SERVER_START_FAILED, + {}, + "Failed to schedule MetaTrader file bridge polling task."); + return; + } + + m_task_manager.run(); + notify_status(BridgeStatus::SERVER_STARTED, m_layout.root.u8string()); + } + + /// \brief Processes one bounded polling window. + /// \details This method is public for deterministic tests and manual + /// owner-loop integrations. `run()` calls it periodically. + void process() { + { + std::lock_guard io_lock(m_io_mutex); + // File polling is intentionally single-flight. External + // callbacks run outside this lock, but another poll must not + // claim the same command window while the current one is in + // its idempotency handoff window. + if (m_processing) { + return; + } + m_processing = true; + } + + try { + process_impl(); + } catch (...) { + std::lock_guard io_lock(m_io_mutex); + m_processing = false; + throw; + } + + { + std::lock_guard io_lock(m_io_mutex); + m_processing = false; + } + } + + /// \brief Stops polling and reports bridge shutdown. + void shutdown() override { + bool was_running = false; + { + std::lock_guard io_lock(m_io_mutex); + was_running = m_running; + m_running = false; + } + + m_task_manager.shutdown(); + if (was_running) { + notify_status(BridgeStatus::SERVER_STOPPED); + } + } + + /// \brief Returns the current resolved client root. + std::filesystem::path client_root() const { + std::lock_guard io_lock(m_io_mutex); + return m_layout.root; + } + + private: + static constexpr int jsonrpc_invalid_request = -32600; + static constexpr int jsonrpc_method_not_found = -32601; + static constexpr int jsonrpc_invalid_params = -32602; + static constexpr int jsonrpc_internal_error = -32603; + + mutable std::mutex m_config_mutex; + mutable std::mutex m_io_mutex; + std::shared_ptr m_state; + std::shared_ptr m_config; + utils::TaskManager m_task_manager; + + detail::FileTransportLayout m_layout; + bool m_runtime_initialized = false; + bool m_running = false; + bool m_processing = false; + std::uint64_t m_last_command_file_seq = 0; + std::uint64_t m_command_scan_offset = 0; + bool m_command_first_file_seq_known = false; + std::uint64_t m_command_first_file_seq = 0; + std::uint64_t m_next_event_file_seq = 1; + std::uint64_t m_event_stream_seq = 0; + std::uint64_t m_state_version = 0; + std::string m_event_stream_id; + std::map m_idempotency_records; + std::map m_idempotency_tombstones; + std::map m_request_id_index; + std::uint64_t m_idempotency_processed_through_file_seq = 0; + + std::shared_ptr get_config_or_throw() const { + std::lock_guard lock(m_config_mutex); + if (!m_config) { + throw std::invalid_argument("MetaTrader file bridge is not configured."); + } + return m_config; + } + + signal_id_allocator_t get_signal_id_allocator() const { + std::lock_guard lock(m_state->mutex); + return m_state->signal_id_allocator; + } + + trade_signal_callback_t get_trade_signal_callback() const { + std::lock_guard lock(m_state->mutex); + return m_state->trade_signal_callback; + } + + std::shared_ptr get_account_info_snapshot() const { + std::lock_guard lock(m_state->mutex); + return m_state->account_info; + } + + void process_impl() { + auto config = get_config_or_throw(); + auto allocator = get_signal_id_allocator(); + auto signal_callback = get_trade_signal_callback(); + + detail::NdjsonSequenceReadResult window; + + { + std::lock_guard io_lock(m_io_mutex); + ensure_runtime_started_locked(*config); + window = read_commands_window_locked(*config); + } + + for (const auto& malformed : window.malformed_records) { + notify_signal_report( + detail::make_signal_report( + *config, + BridgeSignalReportStatus::INVALID, + "malformed_ndjson_record", + malformed.message, + nlohmann::json(), + nlohmann::json(), + {}, + {}, + {}, + nlohmann::json{ + {"start_offset", malformed.start_offset}, + {"next_offset", malformed.next_offset} + })); + } + + if (window.records.empty()) { + std::lock_guard io_lock(m_io_mutex); + // Malformed lines have no domain file_seq, so after their + // diagnostic reports are emitted the byte cursor is the only + // progress marker available for this window. + commit_command_scan_offset_locked(window.next_offset); + return; + } + + for (const auto& record : window.records) { + std::vector reports; + std::vector pending_signals; + { + std::lock_guard io_lock(m_io_mutex); + handle_command_record_locked( + *config, + record, + static_cast(allocator), + static_cast(signal_callback), + reports, + pending_signals); + } + + for (const auto& report : reports) { + notify_signal_report(report); + } + + for (auto& pending : pending_signals) { + process_pending_signal_dispatch( + *config, + pending, + allocator, + signal_callback); + } + + { + std::lock_guard io_lock(m_io_mutex); + if (record.file_seq > m_last_command_file_seq) { + m_last_command_file_seq = record.file_seq; + write_commands_checkpoint_locked(*config); + } + // The cursor is a read-ahead optimization, not the source + // of truth. It advances only after the command has either + // produced a durable result or advanced the file_seq + // checkpoint, so an exception cannot skip a trade command. + commit_command_scan_offset_locked(record.next_offset); + } + } + + { + std::lock_guard io_lock(m_io_mutex); + commit_command_scan_offset_locked(window.next_offset); + } + } + + void ensure_runtime_started_locked(const MetaTraderFileBridgeConfig& config) { + if (m_runtime_initialized) { + return; + } + + m_layout = detail::make_layout(config); + detail::ensure_runtime_directories(m_layout); + m_last_command_file_seq = + read_last_file_seq_checkpoint(m_layout.commands_checkpoint(), config.max_line_bytes); + const auto last_event_reader_seq = + read_last_file_seq_checkpoint(m_layout.events_checkpoint(), config.max_line_bytes); + m_next_event_file_seq = detail::next_file_seq_after_checkpoint( + m_layout.events_log(), + last_event_reader_seq, + config.max_line_bytes); + m_command_scan_offset = 0; + m_command_first_file_seq_known = false; + m_command_first_file_seq = 0; + m_event_stream_seq = 0; + m_state_version = 0; + m_event_stream_id = + detail::stream_id(config) + "-" + + std::to_string(detail::unix_time_ms()) + "-" + + std::to_string(std::chrono::steady_clock::now().time_since_epoch().count()); + const auto idempotency_state = + detail::read_idempotency_state( + m_layout.idempotency_state(), + config.max_idempotency_state_bytes); + m_idempotency_records = idempotency_state.records; + m_idempotency_tombstones = idempotency_state.tombstones; + m_request_id_index = idempotency_state.request_index; + m_idempotency_processed_through_file_seq = + idempotency_state.processed_through_file_seq; + m_last_command_file_seq = std::max( + m_last_command_file_seq, + m_idempotency_processed_through_file_seq); + m_runtime_initialized = true; + } + + detail::NdjsonSequenceReadResult read_commands_window_locked( + const MetaTraderFileBridgeConfig& config) { + std::error_code ec; + const auto exists = std::filesystem::exists(m_layout.commands_log(), ec); + if (ec) { + throw std::runtime_error("Failed to inspect commands log: " + ec.message()); + } + if (!exists) { + m_command_scan_offset = 0; + m_command_first_file_seq_known = false; + m_command_first_file_seq = 0; + return {}; + } + + const auto file_size = std::filesystem::file_size(m_layout.commands_log(), ec); + if (ec) { + throw std::runtime_error("Failed to read commands log size: " + ec.message()); + } + if (file_size > config.max_command_log_bytes) { + throw std::runtime_error("MetaTrader file bridge commands.ndjson exceeds configured byte limit."); + } + if (m_command_scan_offset > file_size) { + m_command_scan_offset = 0; + m_command_first_file_seq_known = false; + } + + std::size_t prefix_scanned_records = 0; + bool prefix_reset_scan_offset = false; + if (m_command_scan_offset != 0) { + // MQL owns commands.ndjson and may clear/regrow it. A stale + // byte offset is safe only while the first valid file_seq still + // matches the identity observed by previous polls. The prefix + // scan skips malformed complete lines so a bad first line does + // not force a full history rescan on every poll. + const auto prefix = command_log_prefix_state_locked(config); + prefix_scanned_records = prefix.scanned_records; + if (prefix.first_file_seq) { + if (m_command_first_file_seq_known && + *prefix.first_file_seq != m_command_first_file_seq) { + m_command_scan_offset = 0; + prefix_reset_scan_offset = true; + } + m_command_first_file_seq_known = true; + m_command_first_file_seq = *prefix.first_file_seq; + if (*prefix.first_file_seq > m_last_command_file_seq) { + m_command_scan_offset = 0; + prefix_reset_scan_offset = true; + } + } else if (prefix.saw_complete_line && + m_command_first_file_seq_known) { + m_command_scan_offset = 0; + prefix_reset_scan_offset = true; + m_command_first_file_seq_known = false; + m_command_first_file_seq = 0; + } + } + + auto scan_budget = config.max_scanned_records_per_poll; + if (prefix_scanned_records != 0 && + !prefix_reset_scan_offset && + scan_budget > 1) { + const auto consumed = std::min( + prefix_scanned_records, + scan_budget - 1); + scan_budget -= consumed; + } + auto window = detail::read_ndjson_sequence_window( + m_layout.commands_log(), + m_command_scan_offset, + m_last_command_file_seq, + config.max_line_bytes, + scan_budget, + config.max_returned_records_per_poll); + if (!m_command_first_file_seq_known && !window.records.empty()) { + m_command_first_file_seq_known = true; + m_command_first_file_seq = window.records.front().file_seq; + } + return window; + } + + void commit_command_scan_offset_locked(const std::uint64_t next_offset) { + if (next_offset > m_command_scan_offset) { + m_command_scan_offset = next_offset; + } + } + + struct CommandLogPrefixState { + bool saw_complete_line = false; + std::size_t scanned_records = 0; + std::unique_ptr first_file_seq; + }; + + CommandLogPrefixState command_log_prefix_state_locked( + const MetaTraderFileBridgeConfig& config) const { + auto batch = detail::read_ndjson_from_offset( + m_layout.commands_log(), + 0, + config.max_line_bytes, + 1); + if (batch.records.empty() && + !batch.malformed_records.empty() && + config.max_scanned_records_per_poll > 1) { + batch = detail::read_ndjson_from_offset( + m_layout.commands_log(), + 0, + config.max_line_bytes, + config.max_scanned_records_per_poll); + } + CommandLogPrefixState state; + state.saw_complete_line = + !batch.records.empty() || !batch.malformed_records.empty(); + state.scanned_records = batch.scanned_records; + if (!batch.records.empty()) { + state.first_file_seq = + std::make_unique(batch.records.front().file_seq); + } + return state; + } + + static std::uint64_t read_last_file_seq_checkpoint( + const std::filesystem::path& checkpoint, + const std::size_t max_line_bytes) { + std::error_code ec; + const auto exists = std::filesystem::exists(checkpoint, ec); + if (ec || !exists) { + return 0; + } + + const auto parsed = detail::read_json_file(checkpoint, max_line_bytes); + if (!parsed.is_object() || !parsed.contains("last_file_seq")) { + throw std::runtime_error( + "MetaTrader file bridge checkpoint is missing last_file_seq: " + + checkpoint.u8string()); + } + const auto& value = parsed.at("last_file_seq"); + if (value.is_number_unsigned()) { + return value.get(); + } + if (value.is_number_integer()) { + const auto signed_value = value.get(); + if (signed_value >= 0) { + return static_cast(signed_value); + } + } + throw std::runtime_error( + "MetaTrader file bridge checkpoint last_file_seq must be a non-negative integer: " + + checkpoint.u8string()); + } + + nlohmann::json idempotency_records_document_locked() const { + return detail::make_idempotency_state_document( + m_idempotency_records, + m_idempotency_tombstones, + m_request_id_index, + m_idempotency_processed_through_file_seq); + } + + bool prune_expired_idempotency_tombstones_locked( + const MetaTraderFileBridgeConfig& config, + const std::uint64_t now_ms = + static_cast(detail::unix_time_ms())) { + bool changed = false; + // Tombstones preserve retry/conflict semantics for evicted + // completed operations only within the configured idempotency + // horizon. In-doubt active records are never removed here. + for (auto it = m_idempotency_tombstones.begin(); + it != m_idempotency_tombstones.end();) { + const auto evicted_at = it->second.evicted_at_ms; + const auto age_ms = now_ms >= evicted_at ? now_ms - evicted_at : 0; + if (evicted_at == 0 || age_ms >= config.idempotency_retention_ms) { + it = m_idempotency_tombstones.erase(it); + changed = true; + } else { + ++it; + } + } + return remove_stale_request_index_locked() || changed; + } + + bool remove_stale_request_index_locked() { + bool changed = false; + for (auto it = m_request_id_index.begin(); + it != m_request_id_index.end();) { + if (m_idempotency_records.find(it->second) == m_idempotency_records.end() && + m_idempotency_tombstones.find(it->second) == m_idempotency_tombstones.end()) { + it = m_request_id_index.erase(it); + changed = true; + } else { + ++it; + } + } + return changed; + } + + void erase_oldest_idempotency_record_locked() { + if (m_idempotency_records.empty()) { + return; + } + + auto oldest = m_idempotency_records.end(); + for (auto it = m_idempotency_records.begin(); + it != m_idempotency_records.end(); + ++it) { + if (detail::string_value(it->second.result, "status") == "in_doubt") { + continue; + } + if (oldest == m_idempotency_records.end()) { + oldest = it; + continue; + } + const auto lhs = it->second.updated_at_ms != 0 + ? it->second.updated_at_ms + : it->second.created_at_ms; + const auto rhs = oldest->second.updated_at_ms != 0 + ? oldest->second.updated_at_ms + : oldest->second.created_at_ms; + if (lhs < rhs) { + oldest = it; + } + } + if (oldest == m_idempotency_records.end()) { + // Dropping an in-doubt record could create a duplicate trade + // after restart. Once the registry is full of unresolved + // handoffs, the bridge must fail closed. + throw std::runtime_error( + "MetaTrader file bridge idempotency state is full of in_doubt operations."); + } + m_idempotency_tombstones[oldest->first] = detail::IdempotencyTombstone{ + oldest->second.payload_fingerprint, + oldest->second.result, + static_cast(detail::unix_time_ms()) + }; + m_idempotency_records.erase(oldest); + } + + void prune_idempotency_records_locked(const MetaTraderFileBridgeConfig& config) { + while (m_idempotency_records.size() > config.max_idempotency_records) { + erase_oldest_idempotency_record_locked(); + } + } + + void write_idempotency_records_locked(const MetaTraderFileBridgeConfig& config) { + prune_expired_idempotency_tombstones_locked(config); + prune_idempotency_records_locked(config); + auto document = idempotency_records_document_locked(); + auto text = document.dump(2); + while (text.size() > config.max_idempotency_state_bytes && + m_idempotency_records.size() > 1) { + erase_oldest_idempotency_record_locked(); + prune_expired_idempotency_tombstones_locked(config); + document = idempotency_records_document_locked(); + text = document.dump(2); + } + if (text.size() > config.max_idempotency_state_bytes) { + throw std::runtime_error( + "MetaTrader file bridge idempotency state exceeds configured byte limit."); + } + detail::write_text_file_atomic(m_layout.idempotency_state(), text); + } + + static std::string idempotency_storage_key( + const std::string& method, + const std::string& idempotency_key) { + if (idempotency_key.empty()) { + return {}; + } + return method + "\n" + idempotency_key; + } + + static std::string request_storage_key( + const std::string& method, + const nlohmann::json& id) { + if (id.is_null()) { + return {}; + } + return method + "\n" + id.dump(-1); + } + + static std::string payload_fingerprint(const nlohmann::json& params) { + return params.dump(-1); + } + + static nlohmann::json make_in_doubt_result( + const MetaTraderFileBridgeConfig& config, + const detail::NdjsonRecord& record, + const SignalId signal_id) { + return nlohmann::json{ + {"status", "in_doubt"}, + {"final", false}, + {"operation_id", "file:" + std::to_string(config.bridge_id) + ":" + std::to_string(record.file_seq)}, + {"signal_ref", { + {"signal_id", std::to_string(signal_id)} + }}, + {"reason", { + {"code", "operation_handoff_in_doubt"}, + {"message", "The command was durably marked before handoff; bridge restart must not dispatch it again automatically."} + }} + }; + } + + std::uint64_t next_event_stream_seq_locked() { + if (m_event_stream_seq == std::numeric_limits::max()) { + throw std::overflow_error("MetaTrader file bridge event seq overflow."); + } + return ++m_event_stream_seq; + } + + std::string next_event_id_locked(const char* prefix) { + return std::string("evt-") + + prefix + + "-" + + std::to_string(m_next_event_file_seq) + + "-" + + std::to_string(m_event_stream_seq + 1); + } + + void append_event_locked(nlohmann::json document) { + detail::append_json_line( + m_layout.events_log(), + detail::with_file_seq(std::move(document), m_next_event_file_seq++), + get_config_or_throw()->max_line_bytes); + } + + void write_commands_checkpoint_locked(const MetaTraderFileBridgeConfig& config) { + if (m_last_command_file_seq > m_idempotency_processed_through_file_seq) { + // The high-water mark protects against replay when + // commands.checkpoint.json is lost. Keep memory and disk in + // sync: if the idempotency state write fails, roll back the + // in-memory high-water update too. + const auto records_backup = m_idempotency_records; + const auto tombstones_backup = m_idempotency_tombstones; + const auto request_index_backup = m_request_id_index; + const auto processed_backup = m_idempotency_processed_through_file_seq; + m_idempotency_processed_through_file_seq = m_last_command_file_seq; + try { + write_idempotency_records_locked(config); + } catch (...) { + m_idempotency_records = records_backup; + m_idempotency_tombstones = tombstones_backup; + m_request_id_index = request_index_backup; + m_idempotency_processed_through_file_seq = processed_backup; + throw; + } + } + detail::write_json_file_atomic( + m_layout.commands_checkpoint(), + detail::make_log_checkpoint(m_last_command_file_seq)); + } + + void write_state_snapshot_locked(const MetaTraderFileBridgeConfig& config) { + auto account = get_account_info_snapshot(); + nlohmann::json accounts = nlohmann::json::array(); + if (account) { + accounts.push_back(detail::account_snapshot_json(*account)); + } + + detail::write_json_file_atomic( + m_layout.state_snapshot(), + detail::make_state_snapshot( + ++m_state_version, + detail::unix_time_ms(), + account ? detail::connection_string(*account) : std::string("unknown"), + std::move(accounts), + nlohmann::json::array())); + (void)config; + } + + void notify_status( + BridgeStatus status, + std::string connection_id = {}, + std::string message = {}) const { + bridge_status_callback_t callback; + { + std::lock_guard lock(m_state->mutex); + callback = m_state->status_callback; + } + if (callback) { + callback(BridgeStatusUpdate( + status, + std::move(connection_id), + std::move(message))); + } + } + + void notify_signal_report(const BridgeSignalReport& report) const { + signal_report_callback_t callback; + { + std::lock_guard lock(m_state->mutex); + callback = m_state->signal_report_callback; + } + if (callback) { + callback(report); + } + } + + void handle_command_record_locked( + const MetaTraderFileBridgeConfig& config, + const detail::NdjsonRecord& record, + const bool has_signal_allocator, + const bool has_trade_signal_callback, + std::vector& reports, + std::vector& pending_signals) { + const auto& document = record.document; + const auto id = document.contains("id") ? document.at("id") : nlohmann::json(nullptr); + try { + if (!document.is_object() || + document.value("jsonrpc", std::string()) != "2.0" || + !document.contains("method") || + !document.at("method").is_string()) { + append_rpc_error_locked( + id, + jsonrpc_invalid_request, + "Invalid JSON-RPC request."); + reports.push_back(detail::make_signal_report( + config, + BridgeSignalReportStatus::INVALID, + "invalid_jsonrpc_request", + "MetaTrader file command is not a valid JSON-RPC request.", + document)); + return; + } + + const auto method = document.at("method").get(); + const auto params = + document.contains("params") ? document.at("params") : nlohmann::json::object(); + + if (method == "protocol.hello") { + append_rpc_result_locked( + id, + nlohmann::json{ + {"status", "ok"}, + {"protocol_versions", nlohmann::json::array({"1.0"})}, + {"supported_methods", nlohmann::json::array({ + "protocol.hello", + "account.balance.get", + "signal.submit", + "trade.open" + })}, + {"transport", "metatrader_file"} + }); + return; + } + + if (method == "account.balance.get") { + handle_balance_get_locked(id); + return; + } + + if (method == "signal.submit" || method == "trade.open") { + handle_signal_command_locked( + config, + record, + id, + method, + params, + has_signal_allocator, + has_trade_signal_callback, + reports, + pending_signals); + return; + } + + append_rpc_error_locked( + id, + jsonrpc_method_not_found, + "Unsupported MetaTrader file bridge method.", + nlohmann::json{{"method", method}}); + reports.push_back(detail::make_signal_report( + config, + BridgeSignalReportStatus::IGNORED, + "unsupported_method", + "MetaTrader file command method is unsupported.", + document, + params, + detail::json_id_to_string(id))); + } catch (const std::exception& ex) { + append_rpc_error_locked( + id, + jsonrpc_internal_error, + ex.what()); + reports.push_back(detail::make_signal_report( + config, + BridgeSignalReportStatus::INTAKE_ERROR, + "command_processing_error", + ex.what(), + document)); + } + } + + void handle_balance_get_locked(const nlohmann::json& id) { + auto account = get_account_info_snapshot(); + if (!account) { + append_rpc_result_locked( + id, + nlohmann::json{ + {"status", "unavailable"}, + {"final", true}, + {"reason", { + {"code", "account_snapshot_unavailable"}, + {"message", "No account snapshot is available."} + }} + }); + return; + } + + append_rpc_result_locked( + id, + nlohmann::json{ + {"status", "completed"}, + {"final", true}, + {"account", detail::account_snapshot_json(*account)} + }); + } + + void handle_signal_command_locked( + const MetaTraderFileBridgeConfig& config, + const detail::NdjsonRecord& record, + const nlohmann::json& id, + const std::string& method, + const nlohmann::json& params, + const bool has_signal_allocator, + const bool has_trade_signal_callback, + std::vector& reports, + std::vector& pending_signals) { + prune_expired_idempotency_tombstones_locked(config); + const auto rpc_request_key = request_storage_key(method, id); + if (!rpc_request_key.empty()) { + const auto request_it = m_request_id_index.find(rpc_request_key); + if (request_it != m_request_id_index.end()) { + const auto record_it = m_idempotency_records.find(request_it->second); + if (record_it != m_idempotency_records.end()) { + append_rpc_result_locked(id, record_it->second.result); + return; + } + const auto tombstone_it = + m_idempotency_tombstones.find(request_it->second); + if (tombstone_it != m_idempotency_tombstones.end()) { + append_rpc_result_locked(id, tombstone_it->second.result); + return; + } + m_request_id_index.erase(request_it); + } + } + + const auto idempotency_key = detail::context_idempotency_key(params); + if (idempotency_key.empty()) { + append_rpc_error_locked( + id, + jsonrpc_invalid_params, + "MetaTrader file bridge trade-affecting commands require context.idempotency_key."); + reports.push_back(detail::make_signal_report( + config, + BridgeSignalReportStatus::INVALID, + "missing_idempotency_key", + "MetaTrader file bridge trade-affecting commands require context.idempotency_key.", + record.document, + params, + detail::json_id_to_string(id))); + return; + } + + const auto storage_key = idempotency_storage_key(method, idempotency_key); + const auto fingerprint = payload_fingerprint(params); + + std::unique_ptr signal; + try { + signal = detail::parse_signal_params(params, method == "trade.open"); + } catch (const std::exception& ex) { + append_rpc_error_locked( + id, + jsonrpc_invalid_params, + ex.what()); + reports.push_back(detail::make_signal_report( + config, + BridgeSignalReportStatus::INVALID, + "invalid_params", + ex.what(), + record.document, + params, + detail::json_id_to_string(id))); + return; + } + + const auto existing = m_idempotency_records.find(storage_key); + if (existing != m_idempotency_records.end()) { + if (existing->second.payload_fingerprint == fingerprint) { + if (!rpc_request_key.empty()) { + m_request_id_index[rpc_request_key] = storage_key; + } + append_rpc_result_locked(id, existing->second.result); + return; + } + + const nlohmann::json conflict = { + {"status", "rejected"}, + {"final", true}, + {"reason", { + {"code", "idempotency_conflict"}, + {"message", "The same idempotency_key was used with a different payload."} + }} + }; + append_rpc_result_locked(id, conflict); + reports.push_back(detail::make_signal_report( + config, + BridgeSignalReportStatus::REJECTED, + "idempotency_conflict", + "MetaTrader file command idempotency key conflicts with an earlier payload.", + record.document, + params, + detail::json_id_to_string(id), + idempotency_key, + detail::clone_candidate_signal(signal))); + return; + } + + const auto tombstone = m_idempotency_tombstones.find(storage_key); + if (tombstone != m_idempotency_tombstones.end()) { + if (tombstone->second.payload_fingerprint == fingerprint) { + if (!rpc_request_key.empty()) { + m_request_id_index[rpc_request_key] = storage_key; + } + append_rpc_result_locked(id, tombstone->second.result); + return; + } + + const nlohmann::json conflict = { + {"status", "rejected"}, + {"final", true}, + {"reason", { + {"code", "idempotency_conflict"}, + {"message", "The same idempotency_key was used with a different payload."} + }} + }; + append_rpc_result_locked(id, conflict); + reports.push_back(detail::make_signal_report( + config, + BridgeSignalReportStatus::REJECTED, + "idempotency_conflict", + "MetaTrader file command idempotency tombstone conflicts with a new payload.", + record.document, + params, + detail::json_id_to_string(id), + idempotency_key, + detail::clone_candidate_signal(signal))); + return; + } + + signal->bridge_id = config.bridge_id; + const auto& context = detail::object_member_or_empty(params, "context"); + if (!context.contains("valid_until_ms")) { + append_rpc_error_locked( + id, + jsonrpc_invalid_params, + "MetaTrader file bridge trade-affecting commands require context.valid_until_ms."); + reports.push_back(detail::make_signal_report( + config, + BridgeSignalReportStatus::INVALID, + "missing_valid_until_ms", + "MetaTrader file bridge trade-affecting commands require context.valid_until_ms.", + record.document, + params, + detail::json_id_to_string(id), + idempotency_key, + detail::clone_candidate_signal(signal))); + return; + } + + std::int64_t valid_until_ms = 0; + try { + valid_until_ms = detail::context_valid_until_ms(params); + } catch (const std::exception& ex) { + append_rpc_error_locked( + id, + jsonrpc_invalid_params, + ex.what()); + reports.push_back(detail::make_signal_report( + config, + BridgeSignalReportStatus::INVALID, + "invalid_valid_until_ms", + ex.what(), + record.document, + params, + detail::json_id_to_string(id), + idempotency_key, + detail::clone_candidate_signal(signal))); + return; + } + if (valid_until_ms <= 0) { + append_rpc_error_locked( + id, + jsonrpc_invalid_params, + "MetaTrader file bridge context.valid_until_ms must be a positive Unix millisecond timestamp."); + reports.push_back(detail::make_signal_report( + config, + BridgeSignalReportStatus::INVALID, + "invalid_valid_until_ms", + "MetaTrader file bridge context.valid_until_ms must be a positive Unix millisecond timestamp.", + record.document, + params, + detail::json_id_to_string(id), + idempotency_key, + detail::clone_candidate_signal(signal))); + return; + } + if (valid_until_ms > 0 && detail::unix_time_ms() > valid_until_ms) { + const nlohmann::json result = { + {"status", "rejected"}, + {"final", true}, + {"reason", { + {"code", "stale_request"}, + {"message", "Command valid_until_ms is in the past."} + }} + }; + append_rpc_result_locked(id, result); + store_idempotency_result_locked( + config, + storage_key, + rpc_request_key, + fingerprint, + result); + reports.push_back(detail::make_signal_report( + config, + BridgeSignalReportStatus::REJECTED, + "stale_request", + "MetaTrader file command expired before processing.", + record.document, + params, + detail::json_id_to_string(id), + signal->unique_hash, + detail::clone_candidate_signal(signal))); + return; + } + + if (!has_signal_allocator) { + append_rpc_error_locked( + id, + jsonrpc_internal_error, + "MetaTrader file bridge signal ID allocator is not configured."); + reports.push_back(detail::make_signal_report( + config, + BridgeSignalReportStatus::INTAKE_ERROR, + "missing_signal_id_allocator", + "MetaTrader file bridge signal ID allocator is not configured.", + record.document, + params, + detail::json_id_to_string(id), + signal->unique_hash, + detail::clone_candidate_signal(signal))); + return; + } + + if (!has_trade_signal_callback) { + const nlohmann::json result = { + {"status", "rejected"}, + {"final", true}, + {"reason", { + {"code", "signal_handler_unavailable"}, + {"message", "MetaTrader file bridge trade signal callback is not configured."} + }} + }; + append_rpc_result_locked(id, result); + store_idempotency_result_locked( + config, + storage_key, + rpc_request_key, + fingerprint, + result); + reports.push_back(detail::make_signal_report( + config, + BridgeSignalReportStatus::REJECTED, + "signal_handler_unavailable", + "MetaTrader file bridge trade signal callback is not configured.", + record.document, + params, + detail::json_id_to_string(id), + signal->unique_hash, + detail::clone_candidate_signal(signal))); + return; + } + + pending_signals.push_back(PendingSignalDispatch{ + record, + id, + params, + std::move(signal), + idempotency_key, + storage_key, + rpc_request_key, + fingerprint + }); + } + + void process_pending_signal_dispatch( + const MetaTraderFileBridgeConfig& config, + PendingSignalDispatch& pending, + const signal_id_allocator_t& allocator, + const trade_signal_callback_t& callback) { + try { + pending.signal->signal_id = allocator(); + } catch (const std::exception& ex) { + append_pending_signal_error( + config, + pending, + jsonrpc_internal_error, + "MetaTrader file bridge signal ID allocator failed.", + "signal_id_allocation_exception", + ex.what()); + return; + } + + if (pending.signal->signal_id == 0) { + append_pending_signal_error( + config, + pending, + jsonrpc_internal_error, + "MetaTrader file bridge could not allocate signal ID.", + "signal_id_allocation_failed", + "MetaTrader file bridge could not allocate signal ID."); + return; + } + + auto candidate = detail::clone_candidate_signal(pending.signal); + { + std::lock_guard io_lock(m_io_mutex); + // Persist the handoff marker before invoking user code. If the + // process dies during or after the callback, restart returns + // the in_doubt result instead of dispatching the trade again. + store_idempotency_result_locked( + config, + pending.idempotency_storage_key, + pending.request_storage_key, + pending.payload_fingerprint, + make_in_doubt_result(config, pending.record, candidate->signal_id)); + } + + try { + callback(std::move(pending.signal)); + } catch (const std::exception& ex) { + append_pending_signal_error( + config, + pending, + jsonrpc_internal_error, + ex.what(), + "signal_callback_exception", + ex.what(), + candidate); + return; + } catch (...) { + append_pending_signal_error( + config, + pending, + jsonrpc_internal_error, + "MetaTrader file bridge signal callback failed with an unknown exception.", + "signal_callback_exception", + "MetaTrader file bridge signal callback failed with an unknown exception.", + candidate); + return; + } + + const nlohmann::json result = { + {"status", "accepted"}, + {"final", false}, + {"operation_id", "file:" + std::to_string(config.bridge_id) + ":" + std::to_string(pending.record.file_seq)}, + {"signal_ref", { + {"signal_id", std::to_string(candidate->signal_id)} + }} + }; + + { + std::lock_guard io_lock(m_io_mutex); + store_idempotency_result_locked( + config, + pending.idempotency_storage_key, + pending.request_storage_key, + pending.payload_fingerprint, + result); + append_rpc_result_locked(pending.id, result); + } + } + + void append_pending_signal_error( + const MetaTraderFileBridgeConfig& config, + const PendingSignalDispatch& pending, + const int error_code, + const std::string& rpc_message, + std::string report_code, + std::string report_message, + std::shared_ptr candidate = {}) { + { + std::lock_guard io_lock(m_io_mutex); + append_rpc_error_locked( + pending.id, + error_code, + rpc_message); + } + + if (!candidate) { + candidate = detail::clone_candidate_signal(pending.signal); + } + notify_signal_report(detail::make_signal_report( + config, + BridgeSignalReportStatus::INTAKE_ERROR, + std::move(report_code), + std::move(report_message), + pending.record.document, + pending.params, + detail::json_id_to_string(pending.id), + pending.signal ? pending.signal->unique_hash : std::string(), + std::move(candidate))); + } + + void store_idempotency_result_locked( + const MetaTraderFileBridgeConfig& config, + const std::string& storage_key, + const std::string& request_key, + const std::string& fingerprint, + const nlohmann::json& result) { + if (storage_key.empty()) { + return; + } + // State mutations are copy-on-write from the caller's perspective. + // A failed atomic file write must not leave transient records in + // memory, otherwise a later poll could observe state that never + // survived a restart. + const auto records_backup = m_idempotency_records; + const auto tombstones_backup = m_idempotency_tombstones; + const auto request_index_backup = m_request_id_index; + const auto processed_backup = m_idempotency_processed_through_file_seq; + if (!request_key.empty()) { + m_request_id_index[request_key] = storage_key; + } + const auto now = static_cast(detail::unix_time_ms()); + auto existing = m_idempotency_records.find(storage_key); + if (existing == m_idempotency_records.end()) { + m_idempotency_records[storage_key] = detail::IdempotencyRecord{ + fingerprint, + result, + now, + now + }; + } else { + existing->second.payload_fingerprint = fingerprint; + existing->second.result = result; + if (existing->second.created_at_ms == 0) { + existing->second.created_at_ms = now; + } + existing->second.updated_at_ms = now; + } + try { + write_idempotency_records_locked(config); + } catch (...) { + m_idempotency_records = records_backup; + m_idempotency_tombstones = tombstones_backup; + m_request_id_index = request_index_backup; + m_idempotency_processed_through_file_seq = processed_backup; + throw; + } + } + + void append_rpc_result_locked(nlohmann::json id, nlohmann::json result) { + append_event_locked(detail::make_jsonrpc_result(std::move(id), std::move(result))); + } + + void append_rpc_error_locked( + nlohmann::json id, + const int code, + std::string message, + nlohmann::json data = nlohmann::json()) { + append_event_locked( + detail::make_jsonrpc_error( + std::move(id), + code, + std::move(message), + std::move(data))); + } + + }; + +} // namespace optionx::bridges::metatrader_file + +#endif // OPTIONX_HEADER_BRIDGES_METATRADER_FILE_META_TRADER_FILE_BRIDGE_HPP_INCLUDED diff --git a/include/optionx_cpp/bridges/metatrader_file/MetaTraderFileBridgeConfig.hpp b/include/optionx_cpp/bridges/metatrader_file/MetaTraderFileBridgeConfig.hpp index 8f551a7..3920685 100644 --- a/include/optionx_cpp/bridges/metatrader_file/MetaTraderFileBridgeConfig.hpp +++ b/include/optionx_cpp/bridges/metatrader_file/MetaTraderFileBridgeConfig.hpp @@ -22,6 +22,12 @@ namespace optionx::bridges::metatrader_file { {"client_secret", client_secret}, {"poll_interval_ms", poll_interval_ms}, {"max_line_bytes", max_line_bytes}, + {"max_command_log_bytes", max_command_log_bytes}, + {"max_scanned_records_per_poll", max_scanned_records_per_poll}, + {"max_returned_records_per_poll", max_returned_records_per_poll}, + {"max_idempotency_state_bytes", max_idempotency_state_bytes}, + {"max_idempotency_records", max_idempotency_records}, + {"idempotency_retention_ms", idempotency_retention_ms}, {"enable_events", enable_events}, {"enable_state_snapshot", enable_state_snapshot} }; @@ -51,6 +57,29 @@ namespace optionx::bridges::metatrader_file { if (j.contains("max_line_bytes")) { max_line_bytes = j.at("max_line_bytes").get(); } + if (j.contains("max_command_log_bytes")) { + max_command_log_bytes = j.at("max_command_log_bytes").get(); + } + if (j.contains("max_scanned_records_per_poll")) { + max_scanned_records_per_poll = + j.at("max_scanned_records_per_poll").get(); + } + if (j.contains("max_returned_records_per_poll")) { + max_returned_records_per_poll = + j.at("max_returned_records_per_poll").get(); + } + if (j.contains("max_idempotency_state_bytes")) { + max_idempotency_state_bytes = + j.at("max_idempotency_state_bytes").get(); + } + if (j.contains("max_idempotency_records")) { + max_idempotency_records = + j.at("max_idempotency_records").get(); + } + if (j.contains("idempotency_retention_ms")) { + idempotency_retention_ms = + j.at("idempotency_retention_ms").get(); + } if (j.contains("enable_events")) { enable_events = j.at("enable_events").get(); } @@ -89,6 +118,42 @@ namespace optionx::bridges::metatrader_file { if (max_line_bytes == 0) { return {false, "MetaTrader file bridge max_line_bytes must be positive."}; } + if (max_command_log_bytes < max_line_bytes) { + return { + false, + "MetaTrader file bridge max_command_log_bytes must be at least max_line_bytes." + }; + } + if (max_scanned_records_per_poll < 2) { + return { + false, + "MetaTrader file bridge max_scanned_records_per_poll must be at least 2." + }; + } + if (max_returned_records_per_poll == 0) { + return { + false, + "MetaTrader file bridge max_returned_records_per_poll must be positive." + }; + } + if (max_idempotency_state_bytes < max_line_bytes) { + return { + false, + "MetaTrader file bridge max_idempotency_state_bytes must be at least max_line_bytes." + }; + } + if (max_idempotency_records == 0) { + return { + false, + "MetaTrader file bridge max_idempotency_records must be positive." + }; + } + if (idempotency_retention_ms == 0) { + return { + false, + "MetaTrader file bridge idempotency_retention_ms must be positive." + }; + } const auto root = std::filesystem::u8path(common_files_root); if (!path_is_within_or_equal(root, client_root())) { return { @@ -133,6 +198,12 @@ namespace optionx::bridges::metatrader_file { std::string client_secret; ///< Optional directory-level shared secret metadata. std::int64_t poll_interval_ms = 250; ///< Recommended bridge polling interval. std::size_t max_line_bytes = 64 * 1024; ///< Maximum complete NDJSON line size. + std::size_t max_command_log_bytes = 8 * 1024 * 1024; ///< Maximum commands.ndjson size scanned by bridge. + std::size_t max_scanned_records_per_poll = 256; ///< Max complete lines scanned per poll; minimum is 2. + std::size_t max_returned_records_per_poll = 64; ///< Max valid commands handled per poll. + std::size_t max_idempotency_state_bytes = 1024 * 1024; ///< Maximum idempotency state file size. + std::size_t max_idempotency_records = 4096; ///< Maximum retained idempotency records. + std::uint64_t idempotency_retention_ms = 24ull * 60ull * 60ull * 1000ull; ///< Completed-operation dedupe retention window. bool enable_events = true; ///< Append bridge events to events.ndjson. bool enable_state_snapshot = true; ///< Atomically publish state.json snapshots. }; diff --git a/include/optionx_cpp/bridges/metatrader_file/detail/MetaTraderFileBridgeUtils.hpp b/include/optionx_cpp/bridges/metatrader_file/detail/MetaTraderFileBridgeUtils.hpp new file mode 100644 index 0000000..ffddaaf --- /dev/null +++ b/include/optionx_cpp/bridges/metatrader_file/detail/MetaTraderFileBridgeUtils.hpp @@ -0,0 +1,446 @@ +#pragma once +#ifndef OPTIONX_HEADER_BRIDGES_METATRADER_FILE_DETAIL_META_TRADER_FILE_BRIDGE_UTILS_HPP_INCLUDED +#define OPTIONX_HEADER_BRIDGES_METATRADER_FILE_DETAIL_META_TRADER_FILE_BRIDGE_UTILS_HPP_INCLUDED + +/// \file MetaTraderFileBridgeUtils.hpp +/// \brief Defines internal helpers for the MetaTrader file bridge runtime. +/// +/// These helpers belong to the bridge implementation layer. They are included +/// through `bridges/metatrader_file.hpp` and are not a standalone public include +/// entry point. + +namespace optionx::bridges::metatrader_file::detail { + + /// \brief Returns whether an account update should produce balance events. + inline bool should_emit_balance_update(const AccountUpdateStatus status) noexcept { + switch (status) { + case AccountUpdateStatus::CONNECTED: + case AccountUpdateStatus::BALANCE_UPDATED: + case AccountUpdateStatus::ACCOUNT_TYPE_CHANGED: + case AccountUpdateStatus::CURRENCY_CHANGED: + return true; + default: + return false; + } + } + + /// \brief Builds the protocol event source URI for this bridge instance. + inline std::string source_uri(const MetaTraderFileBridgeConfig& config) { + return "optionx://bridge/metatrader_file/" + std::to_string(config.bridge_id); + } + + /// \brief Builds the event stream identity for this bridge/client pair. + inline std::string stream_id(const MetaTraderFileBridgeConfig& config) { + return "metatrader-file-" + + std::to_string(config.bridge_id) + + "-" + + config.client_id; + } + + /// \brief Reads the best available account ID as an opaque protocol string. + inline std::string account_id_string(const BaseAccountInfoData& account) { + try { + auto id = account.get_info(AccountInfoType::USER_ID); + if (!id.empty()) { + return id; + } + } catch (...) { + } + try { + const auto id = account.get_info(AccountInfoType::USER_ID); + if (id != 0) { + return std::to_string(id); + } + } catch (...) { + } + return "0"; + } + + /// \brief Reads the account balance or returns the transport default. + inline double safe_account_balance(const BaseAccountInfoData& account) { + try { + return account.get_info(AccountInfoType::BALANCE); + } catch (...) { + return 0.0; + } + } + + /// \brief Reads the account currency or returns `CurrencyType::UNKNOWN`. + inline CurrencyType safe_account_currency(const BaseAccountInfoData& account) { + try { + return account.get_info(AccountInfoType::CURRENCY); + } catch (...) { + return CurrencyType::UNKNOWN; + } + } + + /// \brief Reads the account type or returns `AccountType::UNKNOWN`. + inline AccountType safe_account_type(const BaseAccountInfoData& account) { + try { + return account.get_info(AccountInfoType::ACCOUNT_TYPE); + } catch (...) { + return AccountType::UNKNOWN; + } + } + + /// \brief Reads the connection state as a protocol string. + inline std::string connection_string(const BaseAccountInfoData& account) { + try { + return account.get_info(AccountInfoType::CONNECTION_STATUS) + ? "connected" + : "disconnected"; + } catch (...) { + return "unknown"; + } + } + + /// \brief Converts an account snapshot into the bridge protocol shape. + inline nlohmann::json account_snapshot_json(const BaseAccountInfoData& account) { + nlohmann::json snapshot = { + {"account_id", account_id_string(account)}, + {"connection", connection_string(account)}, + {"balance", make_money_value( + safe_account_balance(account), + safe_account_currency(account), + 2)} + }; + const auto account_type = safe_account_type(account); + if (account_type != AccountType::UNKNOWN) { + snapshot["account_type"] = to_str(account_type); + } + return snapshot; + } + + /// \brief Clones a pending signal for immutable diagnostics. + inline std::shared_ptr clone_candidate_signal( + const std::unique_ptr& signal) { + if (!signal) { + return {}; + } + return std::shared_ptr(signal->clone()); + } + + /// \brief Builds a standard bridge signal report from file-transport data. + inline BridgeSignalReport make_signal_report( + const MetaTraderFileBridgeConfig& config, + BridgeSignalReportStatus status, + std::string reason_code, + std::string message, + nlohmann::json raw_payload = nlohmann::json(), + nlohmann::json parsed_payload = nlohmann::json(), + std::string event_id = {}, + std::string dedupe_key = {}, + std::shared_ptr candidate_signal = {}, + nlohmann::json context = nlohmann::json::object()) { + BridgeSignalReport report; + report.bridge_id = config.bridge_id; + report.bridge_type = config.bridge_type(); + report.status = status; + report.reason_code = std::move(reason_code); + report.message = std::move(message); + report.raw_payload = std::move(raw_payload); + report.parsed_payload = std::move(parsed_payload); + report.event_id = std::move(event_id); + report.dedupe_key = std::move(dedupe_key); + report.candidate_signal = std::move(candidate_signal); + report.context = std::move(context); + report.received_time_ms = unix_time_ms(); + if (report.candidate_signal) { + report.symbol = report.candidate_signal->symbol; + report.signal_name = report.candidate_signal->signal_name; + if (report.dedupe_key.empty()) { + report.dedupe_key = report.candidate_signal->unique_hash; + } + } + return report; + } + + /// \brief Formats any JSON-RPC ID as an opaque diagnostic string. + inline std::string json_id_to_string(const nlohmann::json& id) { + if (id.is_string()) { + return id.get(); + } + if (id.is_null()) { + return {}; + } + return id.dump(-1); + } + + /// \brief Returns a nested object when present, otherwise the input object. + inline const nlohmann::json& object_member_or_self( + const nlohmann::json& object, + const char* key) { + if (object.is_object() && + object.contains(key) && + object.at(key).is_object()) { + return object.at(key); + } + return object; + } + + /// \brief Returns a nested object when present, otherwise an empty object. + inline const nlohmann::json& object_member_or_empty( + const nlohmann::json& object, + const char* key) { + static const nlohmann::json empty = nlohmann::json::object(); + if (object.is_object() && + object.contains(key) && + object.at(key).is_object()) { + return object.at(key); + } + return empty; + } + + /// \brief Reads a string-like JSON member. + inline std::string string_value( + const nlohmann::json& object, + const char* key, + std::string fallback = {}) { + if (!object.is_object() || !object.contains(key)) { + return fallback; + } + const auto& value = object.at(key); + if (value.is_string()) { + return value.get(); + } + if (value.is_number_integer()) { + return std::to_string(value.get()); + } + if (value.is_number_unsigned()) { + return std::to_string(value.get()); + } + return fallback; + } + + /// \brief Reads a decimal-like JSON member. + inline double double_value( + const nlohmann::json& object, + const char* key, + const double fallback = 0.0) { + if (!object.is_object() || !object.contains(key)) { + return fallback; + } + const auto& value = object.at(key); + if (value.is_number()) { + return value.get(); + } + if (value.is_string()) { + return std::stod(value.get()); + } + return fallback; + } + + /// \brief Reads a signed integer-like JSON member. + inline std::int64_t int64_value( + const nlohmann::json& object, + const char* key, + const std::int64_t fallback = 0) { + if (!object.is_object() || !object.contains(key)) { + return fallback; + } + const auto& value = object.at(key); + if (value.is_number_integer()) { + return value.get(); + } + if (value.is_number_unsigned()) { + return static_cast(value.get()); + } + if (value.is_string()) { + return std::stoll(value.get()); + } + return fallback; + } + + /// \brief Reads an unsigned integer-like JSON member. + inline std::uint64_t uint64_value( + const nlohmann::json& object, + const char* key, + const std::uint64_t fallback = 0) { + if (!object.is_object() || !object.contains(key)) { + return fallback; + } + const auto& value = object.at(key); + if (value.is_number_unsigned()) { + return value.get(); + } + if (value.is_number_integer()) { + const auto signed_value = value.get(); + return signed_value >= 0 ? static_cast(signed_value) : fallback; + } + if (value.is_string()) { + const auto parsed = std::stoll(value.get()); + return parsed >= 0 ? static_cast(parsed) : fallback; + } + return fallback; + } + + /// \brief Converts optional protocol text to a currency enum. + inline CurrencyType currency_value(const std::string& value) { + if (value.empty()) { + return CurrencyType::UNKNOWN; + } + return to_enum(value); + } + + /// \brief Converts optional protocol text to an account type enum. + inline AccountType account_type_value(const std::string& value) { + if (value.empty()) { + return AccountType::UNKNOWN; + } + return to_enum(value); + } + + /// \brief Converts optional protocol text to an option type enum. + inline OptionType option_type_value(const std::string& value) { + if (value.empty()) { + return OptionType::UNKNOWN; + } + return to_enum(value); + } + + /// \brief Converts optional protocol text to an order type enum. + inline OrderType order_type_value(const std::string& value) { + if (value.empty()) { + return OrderType::UNKNOWN; + } + return to_enum(value); + } + + /// \brief Applies amount and optional currency fields to a signal. + inline void apply_amount(TradeSignal& signal, const nlohmann::json& trade) { + if (!trade.is_object() || !trade.contains("amount")) { + return; + } + const auto& amount = trade.at("amount"); + if (amount.is_object()) { + signal.amount = double_value(amount, "value", signal.amount); + const auto currency = string_value(amount, "currency"); + if (!currency.empty()) { + signal.currency = currency_value(currency); + } + } else if (amount.is_number()) { + signal.amount = amount.get(); + } else if (amount.is_string()) { + signal.amount = std::stod(amount.get()); + } + } + + /// \brief Applies duration or absolute expiration fields to a signal. + inline void apply_expiry(TradeSignal& signal, const nlohmann::json& trade) { + if (trade.is_object() && trade.contains("expiry") && trade.at("expiry").is_object()) { + const auto& expiry = trade.at("expiry"); + const auto kind = string_value(expiry, "kind"); + if (kind == "duration" || expiry.contains("duration_ms")) { + const auto duration_ms = int64_value(expiry, "duration_ms", 0); + if (duration_ms > 0) { + signal.duration = static_cast(duration_ms / 1000); + } + } else if (kind == "absolute" || expiry.contains("expires_at_ms")) { + const auto expires_at_ms = int64_value(expiry, "expires_at_ms", 0); + if (expires_at_ms > 0) { + signal.expiry_time = expires_at_ms / 1000; + } + } + } + const auto duration_ms = int64_value(trade, "duration_ms", 0); + if (duration_ms > 0) { + signal.duration = static_cast(duration_ms / 1000); + } + const auto duration = int64_value(trade, "duration", 0); + if (duration > 0) { + signal.duration = static_cast(duration); + } + const auto duration_sec = int64_value(trade, "duration_sec", 0); + if (duration_sec > 0) { + signal.duration = static_cast(duration_sec); + } + const auto expires_at_ms = int64_value(trade, "expires_at_ms", 0); + if (expires_at_ms > 0) { + signal.expiry_time = expires_at_ms / 1000; + } + const auto expiry_time = int64_value(trade, "expiry_time", 0); + if (expiry_time > 0) { + signal.expiry_time = expiry_time; + } + } + + /// \brief Applies routing selector fields to a signal. + inline void apply_routing(TradeSignal& signal, const nlohmann::json& params) { + const auto& routing = object_member_or_empty(params, "routing"); + const auto& selector = object_member_or_empty(routing, "selector"); + const auto account_id = int64_value(selector, "account_id", 0); + if (account_id != 0) { + signal.account_id = account_id; + } + } + + /// \brief Reads the optional command deadline from context. + inline std::int64_t context_valid_until_ms(const nlohmann::json& params) { + const auto& context = object_member_or_empty(params, "context"); + return int64_value(context, "valid_until_ms", 0); + } + + /// \brief Reads the optional command idempotency key from context. + inline std::string context_idempotency_key(const nlohmann::json& params) { + const auto& context = object_member_or_empty(params, "context"); + return string_value(context, "idempotency_key"); + } + + /// \brief Converts `signal.submit` or `trade.open` params into a TradeSignal. + inline std::unique_ptr parse_signal_params( + const nlohmann::json& params, + const bool direct_trade_open) { + if (!params.is_object()) { + throw std::invalid_argument("Command params must be an object."); + } + + const auto& trade = direct_trade_open + ? object_member_or_self(params, "trade") + : object_member_or_self(params, "signal"); + const auto& identity = object_member_or_empty(params, "identity"); + const auto& context = object_member_or_empty(params, "context"); + + auto signal = std::make_unique(); + signal->symbol = string_value(trade, "symbol"); + signal->signal_name = string_value( + identity, + "signal_name", + string_value(trade, "signal_name", direct_trade_open ? "direct_trade_open" : "")); + signal->unique_hash = string_value( + identity, + "unique_hash", + string_value(trade, "unique_hash", string_value(context, "idempotency_key"))); + signal->unique_id = int64_value(identity, "unique_id", int64_value(trade, "unique_id", 0)); + signal->user_data = string_value(identity, "user_data", string_value(trade, "user_data")); + signal->comment = string_value(trade, "comment"); + signal->account_id = int64_value(trade, "account_id", 0); + signal->account_type = account_type_value(string_value(trade, "account_type")); + signal->currency = currency_value(string_value(trade, "currency")); + signal->option_type = option_type_value(string_value(trade, "option_type")); + signal->order_type = order_type_value( + string_value( + trade, + "order_type", + string_value(trade, "direction", string_value(trade, "action")))); + signal->refund = double_value(trade, "refund", 0.0); + signal->min_payout = double_value(trade, "min_payout", 0.0); + + apply_amount(*signal, trade); + apply_expiry(*signal, trade); + apply_routing(*signal, params); + + if (signal->symbol.empty()) { + throw std::invalid_argument("Command trade symbol is required."); + } + if (signal->order_type == OrderType::UNKNOWN) { + throw std::invalid_argument("Command order_type is required."); + } + if (direct_trade_open && signal->amount <= 0.0) { + throw std::invalid_argument("trade.open amount must be positive."); + } + return signal; + } + +} // namespace optionx::bridges::metatrader_file::detail + +#endif // OPTIONX_HEADER_BRIDGES_METATRADER_FILE_DETAIL_META_TRADER_FILE_BRIDGE_UTILS_HPP_INCLUDED diff --git a/include/optionx_cpp/bridges/metatrader_file/detail/MetaTraderFileIdempotencyStore.hpp b/include/optionx_cpp/bridges/metatrader_file/detail/MetaTraderFileIdempotencyStore.hpp new file mode 100644 index 0000000..dfbdf06 --- /dev/null +++ b/include/optionx_cpp/bridges/metatrader_file/detail/MetaTraderFileIdempotencyStore.hpp @@ -0,0 +1,235 @@ +#pragma once +#ifndef OPTIONX_HEADER_BRIDGES_METATRADER_FILE_DETAIL_META_TRADER_FILE_IDEMPOTENCY_STORE_HPP_INCLUDED +#define OPTIONX_HEADER_BRIDGES_METATRADER_FILE_DETAIL_META_TRADER_FILE_IDEMPOTENCY_STORE_HPP_INCLUDED + +/// \file MetaTraderFileIdempotencyStore.hpp +/// \brief Defines durable idempotency-state helpers for the MetaTrader file bridge. + +namespace optionx::bridges::metatrader_file::detail { + + /// \struct IdempotencyRecord + /// \brief Active operation registry entry retained for duplicate detection. + struct IdempotencyRecord { + std::string payload_fingerprint; ///< Canonical request payload fingerprint. + nlohmann::json result; ///< Result to return for duplicate requests. + std::uint64_t created_at_ms = 0; ///< Creation time in Unix milliseconds. + std::uint64_t updated_at_ms = 0; ///< Last update time in Unix milliseconds. + }; + + /// \struct IdempotencyTombstone + /// \brief Compact retained entry for evicted completed operations. + struct IdempotencyTombstone { + std::string payload_fingerprint; ///< Original request payload fingerprint. + nlohmann::json result; ///< Original operation result retained for retry. + std::uint64_t evicted_at_ms = 0; ///< Eviction time in Unix milliseconds. + }; + + /// \struct IdempotencyState + /// \brief Complete durable idempotency registry snapshot. + struct IdempotencyState { + std::map records; ///< Active records by method/key. + std::map tombstones; ///< Evicted completed records. + std::map request_index; ///< JSON-RPC request ID to operation key. + std::uint64_t processed_through_file_seq = 0; ///< Durable high-water command sequence. + }; + + /// \brief Reads a required non-negative integer JSON field. + inline std::uint64_t required_idempotency_uint64_value( + const nlohmann::json& object, + const char* key, + const std::filesystem::path& file, + const bool allow_zero = true) { + const auto it = object.find(key); + if (it == object.end()) { + throw std::runtime_error( + "MetaTrader file bridge idempotency state is missing numeric field '" + + std::string(key) + "': " + file.u8string()); + } + std::uint64_t result = 0; + if (it->is_number_unsigned()) { + result = it->get(); + } else if (it->is_number_integer()) { + const auto value = it->get(); + if (value < 0) { + throw std::runtime_error( + "MetaTrader file bridge idempotency state field '" + + std::string(key) + "' must be non-negative: " + file.u8string()); + } + result = static_cast(value); + } else { + throw std::runtime_error( + "MetaTrader file bridge idempotency state field '" + + std::string(key) + "' must be an integer: " + file.u8string()); + } + if (!allow_zero && result == 0) { + throw std::runtime_error( + "MetaTrader file bridge idempotency state field '" + + std::string(key) + "' must be positive: " + file.u8string()); + } + return result; + } + + /// \brief Reads an idempotency registry snapshot from JSON. + inline IdempotencyState read_idempotency_state( + const std::filesystem::path& file, + const std::size_t max_state_bytes) { + IdempotencyState state; + + std::error_code ec; + const auto exists = std::filesystem::exists(file, ec); + if (ec) { + throw std::runtime_error("Failed to inspect idempotency state: " + ec.message()); + } + if (!exists) { + return state; + } + + const auto parsed = read_json_file(file, max_state_bytes); + if (!parsed.is_object()) { + throw std::runtime_error( + "MetaTrader file bridge idempotency state must be an object: " + + file.u8string()); + } + state.processed_through_file_seq = + required_idempotency_uint64_value( + parsed, + "processed_through_file_seq", + file); + + const auto it = parsed.find("records"); + if (it == parsed.end()) { + return state; + } + if (!it->is_object()) { + throw std::runtime_error( + "MetaTrader file bridge idempotency records must be an object: " + + file.u8string()); + } + + for (auto record_it = it->begin(); record_it != it->end(); ++record_it) { + const auto& value = record_it.value(); + if (!value.is_object() || + !value.contains("payload_fingerprint") || + !value.at("payload_fingerprint").is_string() || + !value.contains("result")) { + throw std::runtime_error( + "MetaTrader file bridge idempotency record is malformed: " + + file.u8string()); + } + + IdempotencyRecord record; + record.payload_fingerprint = value.at("payload_fingerprint").get(); + record.result = value.at("result"); + record.created_at_ms = + required_idempotency_uint64_value( + value, + "created_at_ms", + file, + false); + record.updated_at_ms = + required_idempotency_uint64_value( + value, + "updated_at_ms", + file, + false); + state.records.emplace(record_it.key(), std::move(record)); + } + + const auto tombstones_it = parsed.find("tombstones"); + if (tombstones_it != parsed.end()) { + if (!tombstones_it->is_object()) { + throw std::runtime_error( + "MetaTrader file bridge idempotency tombstones must be an object: " + + file.u8string()); + } + for (auto tombstone_it = tombstones_it->begin(); + tombstone_it != tombstones_it->end(); + ++tombstone_it) { + const auto& value = tombstone_it.value(); + if (!value.is_object() || + !value.contains("payload_fingerprint") || + !value.at("payload_fingerprint").is_string() || + !value.contains("result")) { + throw std::runtime_error( + "MetaTrader file bridge idempotency tombstone is malformed: " + + file.u8string()); + } + + IdempotencyTombstone tombstone; + tombstone.payload_fingerprint = + value.at("payload_fingerprint").get(); + tombstone.result = value.at("result"); + tombstone.evicted_at_ms = + required_idempotency_uint64_value( + value, + "evicted_at_ms", + file, + false); + state.tombstones.emplace(tombstone_it.key(), std::move(tombstone)); + } + } + + const auto index_it = parsed.find("request_index"); + if (index_it != parsed.end()) { + if (!index_it->is_object()) { + throw std::runtime_error( + "MetaTrader file bridge idempotency request_index must be an object: " + + file.u8string()); + } + for (auto request_it = index_it->begin(); + request_it != index_it->end(); + ++request_it) { + if (!request_it.value().is_string()) { + throw std::runtime_error( + "MetaTrader file bridge idempotency request_index entry is malformed: " + + file.u8string()); + } + state.request_index.emplace( + request_it.key(), + request_it.value().get()); + } + } + return state; + } + + /// \brief Builds an idempotency registry JSON document from in-memory maps. + inline nlohmann::json make_idempotency_state_document( + const std::map& records_map, + const std::map& tombstones_map, + const std::map& request_index_map, + const std::uint64_t processed_through_file_seq) { + nlohmann::json records = nlohmann::json::object(); + for (const auto& item : records_map) { + records[item.first] = { + {"payload_fingerprint", item.second.payload_fingerprint}, + {"result", item.second.result}, + {"created_at_ms", item.second.created_at_ms}, + {"updated_at_ms", item.second.updated_at_ms} + }; + } + nlohmann::json tombstones = nlohmann::json::object(); + for (const auto& item : tombstones_map) { + tombstones[item.first] = { + {"payload_fingerprint", item.second.payload_fingerprint}, + {"result", item.second.result}, + {"evicted_at_ms", item.second.evicted_at_ms} + }; + } + nlohmann::json request_index = nlohmann::json::object(); + for (const auto& item : request_index_map) { + if (records_map.find(item.second) != records_map.end() || + tombstones_map.find(item.second) != tombstones_map.end()) { + request_index[item.first] = item.second; + } + } + return nlohmann::json{ + {"processed_through_file_seq", processed_through_file_seq}, + {"records", std::move(records)}, + {"tombstones", std::move(tombstones)}, + {"request_index", std::move(request_index)} + }; + } + +} // namespace optionx::bridges::metatrader_file::detail + +#endif // OPTIONX_HEADER_BRIDGES_METATRADER_FILE_DETAIL_META_TRADER_FILE_IDEMPOTENCY_STORE_HPP_INCLUDED diff --git a/include/optionx_cpp/bridges/metatrader_file/detail/MetaTraderFileOperationKey.hpp b/include/optionx_cpp/bridges/metatrader_file/detail/MetaTraderFileOperationKey.hpp new file mode 100644 index 0000000..f062425 --- /dev/null +++ b/include/optionx_cpp/bridges/metatrader_file/detail/MetaTraderFileOperationKey.hpp @@ -0,0 +1,40 @@ +#pragma once +#ifndef OPTIONX_HEADER_BRIDGES_METATRADER_FILE_DETAIL_META_TRADER_FILE_OPERATION_KEY_HPP_INCLUDED +#define OPTIONX_HEADER_BRIDGES_METATRADER_FILE_DETAIL_META_TRADER_FILE_OPERATION_KEY_HPP_INCLUDED + +/// \file MetaTraderFileOperationKey.hpp +/// \brief Defines compact operation key helpers for MetaTrader file clients. + +namespace optionx::bridges::metatrader_file { + + /// \brief Generates a compact Base36 operation key for file-transport commands. + /// \details The generated value is intended to be created once per logical + /// operation and reused for JSON-RPC `id`, `context.idempotency_key`, or + /// domain `unique_hash` on retries of that same operation. The timestamp + /// prefix keeps keys roughly time-sortable, while the random tail and local + /// counter avoid collisions inside one process. + /// \param prefix Optional readable prefix. Use an empty string to omit it. + /// \param random_tail_length Number of Base36 random characters to append. + /// \return A compact operation key such as `mfb_lz7abc_4fz..._1`. + inline std::string make_compact_operation_key( + const std::string& prefix = "mfb", + const std::size_t random_tail_length = 16) { + static std::atomic counter{0}; + + const auto now_ms = std::chrono::duration_cast( + std::chrono::system_clock::now().time_since_epoch()).count(); + const auto time_part = utils::Base36::encode_int( + static_cast(now_ms)); + const auto random_part = utils::Base36::random_string(random_tail_length); + const auto counter_part = utils::Base36::encode_int( + static_cast(counter++)); + + if (prefix.empty()) { + return time_part + random_part + counter_part; + } + return prefix + "_" + time_part + "_" + random_part + "_" + counter_part; + } + +} // namespace optionx::bridges::metatrader_file + +#endif // OPTIONX_HEADER_BRIDGES_METATRADER_FILE_DETAIL_META_TRADER_FILE_OPERATION_KEY_HPP_INCLUDED diff --git a/include/optionx_cpp/bridges/metatrader_file/MetaTraderFilePathUtils.hpp b/include/optionx_cpp/bridges/metatrader_file/detail/MetaTraderFilePathUtils.hpp similarity index 93% rename from include/optionx_cpp/bridges/metatrader_file/MetaTraderFilePathUtils.hpp rename to include/optionx_cpp/bridges/metatrader_file/detail/MetaTraderFilePathUtils.hpp index 173a056..9bfc0b5 100644 --- a/include/optionx_cpp/bridges/metatrader_file/MetaTraderFilePathUtils.hpp +++ b/include/optionx_cpp/bridges/metatrader_file/detail/MetaTraderFilePathUtils.hpp @@ -1,6 +1,6 @@ #pragma once -#ifndef OPTIONX_HEADER_BRIDGES_METATRADER_FILE_META_TRADER_FILE_PATH_UTILS_HPP_INCLUDED -#define OPTIONX_HEADER_BRIDGES_METATRADER_FILE_META_TRADER_FILE_PATH_UTILS_HPP_INCLUDED +#ifndef OPTIONX_HEADER_BRIDGES_METATRADER_FILE_DETAIL_META_TRADER_FILE_PATH_UTILS_HPP_INCLUDED +#define OPTIONX_HEADER_BRIDGES_METATRADER_FILE_DETAIL_META_TRADER_FILE_PATH_UTILS_HPP_INCLUDED /// \file MetaTraderFilePathUtils.hpp /// \brief Path and identifier helpers for the MetaTrader file transport. @@ -115,4 +115,4 @@ namespace optionx::bridges::metatrader_file { } // namespace optionx::bridges::metatrader_file -#endif // OPTIONX_HEADER_BRIDGES_METATRADER_FILE_META_TRADER_FILE_PATH_UTILS_HPP_INCLUDED +#endif // OPTIONX_HEADER_BRIDGES_METATRADER_FILE_DETAIL_META_TRADER_FILE_PATH_UTILS_HPP_INCLUDED diff --git a/include/optionx_cpp/bridges/metatrader_file/detail/MetaTraderFileProtocol.hpp b/include/optionx_cpp/bridges/metatrader_file/detail/MetaTraderFileProtocol.hpp index 2740e4e..27fd2a9 100644 --- a/include/optionx_cpp/bridges/metatrader_file/detail/MetaTraderFileProtocol.hpp +++ b/include/optionx_cpp/bridges/metatrader_file/detail/MetaTraderFileProtocol.hpp @@ -28,6 +28,7 @@ namespace optionx::bridges::metatrader_file::detail { std::filesystem::path state_snapshot() const { return root / "state.json"; } std::filesystem::path commands_checkpoint() const { return root / "commands.checkpoint.json"; } std::filesystem::path events_checkpoint() const { return root / "events.checkpoint.json"; } + std::filesystem::path idempotency_state() const { return root / "idempotency.json"; } }; /// \struct NdjsonRecord diff --git a/tests/bridge_umbrella_include_test.cpp b/tests/bridge_umbrella_include_test.cpp index 1717a28..085af0f 100644 --- a/tests/bridge_umbrella_include_test.cpp +++ b/tests/bridge_umbrella_include_test.cpp @@ -23,10 +23,12 @@ TEST(BridgeUmbrellaIncludeTest, ExposesTradingViewBridgeFamily) { TEST(BridgeUmbrellaIncludeTest, ExposesMetaTraderFileBridgeFamily) { optionx::bridges::metatrader_file::MetaTraderFileBridgeConfig config; + optionx::bridges::metatrader_file::MetaTraderFileBridge bridge; EXPECT_EQ( config.bridge_type(), optionx::BridgeType::METATRADER_FILE_TRANSPORT); + EXPECT_EQ(bridge.client_root(), std::filesystem::path()); } TEST(BridgeUmbrellaIncludeTest, ExposesSignalReportApi) { diff --git a/tests/metatrader_file_bridge_test.cpp b/tests/metatrader_file_bridge_test.cpp index f6a41cc..443003f 100644 --- a/tests/metatrader_file_bridge_test.cpp +++ b/tests/metatrader_file_bridge_test.cpp @@ -5,8 +5,10 @@ #include #include #include +#include #include #include +#include namespace { @@ -29,6 +31,91 @@ std::filesystem::path make_temp_root() { ("optionx_metatrader_file_bridge_test_" + std::to_string(stamp)); } +class TestAccountInfo final : public optionx::BaseAccountInfoData { +public: + std::int64_t user_id = 42; + double balance = 1234.5; + optionx::CurrencyType currency = optionx::CurrencyType::USD; + optionx::AccountType account_type = optionx::AccountType::DEMO; + bool connected = true; + + std::unique_ptr clone_unique() const override { + return std::make_unique(*this); + } + + std::shared_ptr clone_shared() const override { + return std::make_shared(*this); + } + +private: + bool get_info_bool(const optionx::AccountInfoRequest& request) const override { + return request.type == optionx::AccountInfoType::CONNECTION_STATUS + ? connected + : false; + } + + std::int64_t get_info_int64(const optionx::AccountInfoRequest& request) const override { + switch (request.type) { + case optionx::AccountInfoType::USER_ID: + return user_id; + case optionx::AccountInfoType::CONNECTION_STATUS: + return connected ? 1 : 0; + case optionx::AccountInfoType::ACCOUNT_TYPE: + return static_cast(account_type); + case optionx::AccountInfoType::CURRENCY: + return static_cast(currency); + default: + return 0; + } + } + + double get_info_f64(const optionx::AccountInfoRequest& request) const override { + return request.type == optionx::AccountInfoType::BALANCE ? balance : 0.0; + } + + std::string get_info_str(const optionx::AccountInfoRequest& request) const override { + return request.type == optionx::AccountInfoType::USER_ID + ? std::to_string(user_id) + : std::string(); + } + + optionx::AccountType get_info_account_type( + const optionx::AccountInfoRequest&) const override { + return account_type; + } + + optionx::CurrencyType get_info_currency( + const optionx::AccountInfoRequest&) const override { + return currency; + } +}; + +nlohmann::json make_signal_submit_params( + std::string idempotency_key, + std::string unique_hash, + std::string symbol = "EURUSD", + std::string order_type = "BUY", + std::int64_t valid_until_ms = + optionx::bridges::metatrader_file::detail::unix_time_ms() + 60000) { + return nlohmann::json{ + {"context", { + {"idempotency_key", std::move(idempotency_key)}, + {"valid_until_ms", valid_until_ms} + }}, + {"identity", { + {"unique_hash", std::move(unique_hash)}, + {"signal_name", "noisy_rsi"} + }}, + {"signal", { + {"symbol", std::move(symbol)}, + {"order_type", std::move(order_type)}, + {"option_type", "SPRINT"}, + {"amount", {{"value", "1.25"}, {"currency", "USD"}}}, + {"expiry", {{"kind", "duration"}, {"duration_ms", 60000}}} + }} + }; +} + } // namespace TEST(MetaTraderFileBridgeConfig, RoundTripsJsonAndValidatesSafeClientId) { @@ -39,6 +126,12 @@ TEST(MetaTraderFileBridgeConfig, RoundTripsJsonAndValidatesSafeClientId) { config.client_secret = "local-secret"; config.poll_interval_ms = 100; config.max_line_bytes = 4096; + config.max_command_log_bytes = 128 * 1024; + config.max_scanned_records_per_poll = 32; + config.max_returned_records_per_poll = 8; + config.max_idempotency_state_bytes = 64 * 1024; + config.max_idempotency_records = 128; + config.idempotency_retention_ms = 12 * 60 * 60 * 1000; config.enable_events = true; config.enable_state_snapshot = false; @@ -55,6 +148,12 @@ TEST(MetaTraderFileBridgeConfig, RoundTripsJsonAndValidatesSafeClientId) { EXPECT_EQ(restored.client_secret, "local-secret"); EXPECT_EQ(restored.poll_interval_ms, 100); EXPECT_EQ(restored.max_line_bytes, 4096u); + EXPECT_EQ(restored.max_command_log_bytes, 128u * 1024u); + EXPECT_EQ(restored.max_scanned_records_per_poll, 32u); + EXPECT_EQ(restored.max_returned_records_per_poll, 8u); + EXPECT_EQ(restored.max_idempotency_state_bytes, 64u * 1024u); + EXPECT_EQ(restored.max_idempotency_records, 128u); + EXPECT_EQ(restored.idempotency_retention_ms, 12u * 60u * 60u * 1000u); EXPECT_TRUE(restored.enable_events); EXPECT_FALSE(restored.enable_state_snapshot); EXPECT_TRUE(restored.validate().first); @@ -79,9 +178,41 @@ TEST(MetaTraderFileBridgeConfig, RoundTripsJsonAndValidatesSafeClientId) { EXPECT_FALSE(restored.validate().first); restored.namespace_subdir = "OptionX/Bridge/v1"; + restored.max_scanned_records_per_poll = 0; + EXPECT_FALSE(restored.validate().first); + + restored.max_scanned_records_per_poll = 1; + EXPECT_FALSE(restored.validate().first); + + restored.max_scanned_records_per_poll = 32; + restored.max_returned_records_per_poll = 0; + EXPECT_FALSE(restored.validate().first); + + restored.max_returned_records_per_poll = 8; + restored.idempotency_retention_ms = 0; + EXPECT_FALSE(restored.validate().first); + + restored.idempotency_retention_ms = 12 * 60 * 60 * 1000; EXPECT_TRUE(restored.validate().first); } +TEST(MetaTraderFileOperationKey, GeneratesCompactUniqueKeys) { + namespace mtfile = optionx::bridges::metatrader_file; + + const auto first = mtfile::make_compact_operation_key("mfb", 12); + const auto second = mtfile::make_compact_operation_key("mfb", 12); + + EXPECT_NE(first, second); + EXPECT_EQ(first.rfind("mfb_", 0), 0u); + EXPECT_EQ(second.rfind("mfb_", 0), 0u); + for (const auto ch : first) { + EXPECT_TRUE( + (ch >= '0' && ch <= '9') || + (ch >= 'a' && ch <= 'z') || + ch == '_'); + } +} + TEST(MetaTraderFileBridgeConfig, AcceptsDotRootAndTrailingSeparators) { namespace mtfile = optionx::bridges::metatrader_file; @@ -625,6 +756,1125 @@ TEST(MetaTraderFileProtocol, BuildsBalanceAndTradeUpdateNotifications) { EXPECT_EQ(cancelled_with_reason.at("failure").at("message").get(), "Cancelled by user."); } +TEST(MetaTraderFileBridge, ProcessesSignalSubmitAndWritesResponse) { + namespace mtfile = optionx::bridges::metatrader_file; + namespace protocol = optionx::bridges::metatrader_file::detail; + + const auto root = make_temp_root(); + ScopedPathCleanup cleanup(root); + + mtfile::MetaTraderFileBridgeConfig config; + config.common_files_root = root.u8string(); + config.bridge_id = 31; + config.client_id = "terminal-01"; + config.max_line_bytes = 8192; + config.max_scanned_records_per_poll = 16; + config.max_returned_records_per_poll = 4; + + const auto layout = protocol::make_layout(config); + protocol::ensure_runtime_directories(layout); + + protocol::append_json_line( + layout.commands_log(), + protocol::make_file_jsonrpc_request( + 1, + "cmd-1", + "signal.submit", + nlohmann::json{ + {"context", { + {"idempotency_key", "mql5:terminal-01:bar-1:buy"}, + {"valid_until_ms", protocol::unix_time_ms() + 60000} + }}, + {"routing", { + {"selector", { + {"kind", "account"}, + {"account_id", "7"} + }} + }}, + {"identity", { + {"unique_hash", "signal-hash-1"}, + {"signal_name", "noisy_rsi"} + }}, + {"signal", { + {"symbol", "EURUSD"}, + {"order_type", "BUY"}, + {"option_type", "SPRINT"}, + {"amount", {{"value", "1.25"}, {"currency", "USD"}}}, + {"expiry", {{"kind", "duration"}, {"duration_ms", 60000}}} + }} + }), + config.max_line_bytes); + + mtfile::MetaTraderFileBridge bridge; + ASSERT_TRUE(bridge.configure(std::make_unique(config))); + + optionx::SignalId next_signal_id = 100; + bridge.on_signal_id() = [&]() { + return next_signal_id++; + }; + + std::vector> signals; + bridge.on_trade_signal() = [&](std::unique_ptr signal) { + ASSERT_TRUE(signal); + signals.push_back(std::move(signal)); + }; + + bridge.process(); + + ASSERT_EQ(signals.size(), 1u); + EXPECT_EQ(signals[0]->signal_id, 100u); + EXPECT_EQ(signals[0]->bridge_id, 31u); + EXPECT_EQ(signals[0]->account_id, 7); + EXPECT_EQ(signals[0]->symbol, "EURUSD"); + EXPECT_EQ(signals[0]->order_type, optionx::OrderType::BUY); + EXPECT_EQ(signals[0]->option_type, optionx::OptionType::SPRINT); + EXPECT_DOUBLE_EQ(signals[0]->amount, 1.25); + EXPECT_EQ(signals[0]->currency, optionx::CurrencyType::USD); + EXPECT_EQ(signals[0]->duration, 60u); + EXPECT_EQ(signals[0]->unique_hash, "signal-hash-1"); + EXPECT_EQ(signals[0]->signal_name, "noisy_rsi"); + + const auto checkpoint = protocol::read_json_file( + layout.commands_checkpoint(), + config.max_line_bytes); + EXPECT_EQ(checkpoint.at("last_file_seq").get(), 1u); + + const auto events = protocol::read_ndjson_since_file_seq( + layout.events_log(), + 0, + config.max_line_bytes); + ASSERT_EQ(events.size(), 1u); + EXPECT_EQ(events[0].file_seq, 1u); + EXPECT_EQ(events[0].document.at("id").get(), "cmd-1"); + const auto& result = events[0].document.at("result"); + EXPECT_EQ(result.at("status").get(), "accepted"); + EXPECT_EQ(result.at("operation_id").get(), "file:31:1"); + EXPECT_EQ(result.at("signal_ref").at("signal_id").get(), "100"); +} + +TEST(MetaTraderFileBridge, ReadsCommandsAfterOwnerClearsAndRegrowsLog) { + namespace mtfile = optionx::bridges::metatrader_file; + namespace protocol = optionx::bridges::metatrader_file::detail; + + const auto root = make_temp_root(); + ScopedPathCleanup cleanup(root); + + mtfile::MetaTraderFileBridgeConfig config; + config.common_files_root = root.u8string(); + config.bridge_id = 33; + config.client_id = "terminal-clear-regrow"; + config.max_line_bytes = 65536; + config.max_returned_records_per_poll = 64; + + const auto layout = protocol::make_layout(config); + protocol::ensure_runtime_directories(layout); + + protocol::append_json_line( + layout.commands_log(), + protocol::make_file_jsonrpc_request( + 1, + "cmd-1", + "signal.submit", + make_signal_submit_params("clear-regrow-1", "signal-hash-1")), + config.max_line_bytes); + + mtfile::MetaTraderFileBridge bridge; + ASSERT_TRUE(bridge.configure(std::make_unique(config))); + + optionx::SignalId next_signal_id = 100; + bridge.on_signal_id() = [&]() { + return next_signal_id++; + }; + + std::vector symbols; + bridge.on_trade_signal() = [&](std::unique_ptr signal) { + ASSERT_TRUE(signal); + symbols.push_back(signal->symbol); + }; + + bridge.process(); + ASSERT_EQ(symbols.size(), 1u); + const auto old_size = std::filesystem::file_size(layout.commands_log()); + + protocol::clear_file_atomic(layout.commands_log()); + + for (std::uint64_t seq = 2; seq < 20; ++seq) { + protocol::append_json_line( + layout.commands_log(), + protocol::make_file_jsonrpc_request( + seq, + "noop-" + std::to_string(seq), + "protocol.hello"), + config.max_line_bytes); + } + protocol::append_json_line( + layout.commands_log(), + protocol::make_file_jsonrpc_request( + 20, + "cmd-20", + "signal.submit", + make_signal_submit_params("clear-regrow-20", "signal-hash-20", "GBPUSD")), + config.max_line_bytes); + ASSERT_GE(std::filesystem::file_size(layout.commands_log()), old_size); + + bridge.process(); + ASSERT_EQ(symbols.size(), 2u); + EXPECT_EQ(symbols[1], "GBPUSD"); + + const auto checkpoint = protocol::read_json_file( + layout.commands_checkpoint(), + config.max_line_bytes); + EXPECT_EQ(checkpoint.at("last_file_seq").get(), 20u); +} + +TEST(MetaTraderFileBridge, UsesBoundedScanAfterOwnerClearsAndRegrowsLog) { + namespace mtfile = optionx::bridges::metatrader_file; + namespace protocol = optionx::bridges::metatrader_file::detail; + + const auto root = make_temp_root(); + ScopedPathCleanup cleanup(root); + + mtfile::MetaTraderFileBridgeConfig config; + config.common_files_root = root.u8string(); + config.bridge_id = 34; + config.client_id = "terminal-bounded-regrow"; + config.max_line_bytes = 8192; + config.max_scanned_records_per_poll = 3; + config.max_returned_records_per_poll = 8; + + const auto layout = protocol::make_layout(config); + protocol::ensure_runtime_directories(layout); + + protocol::append_json_line( + layout.commands_log(), + protocol::make_file_jsonrpc_request( + 1, + "cmd-1", + "signal.submit", + make_signal_submit_params("bounded-regrow-1", "bounded-regrow-1")), + config.max_line_bytes); + + mtfile::MetaTraderFileBridge bridge; + ASSERT_TRUE(bridge.configure(std::make_unique(config))); + optionx::SignalId next_signal_id = 100; + bridge.on_signal_id() = [&]() { + return next_signal_id++; + }; + + std::vector symbols; + bridge.on_trade_signal() = [&](std::unique_ptr signal) { + ASSERT_TRUE(signal); + symbols.push_back(signal->symbol); + }; + + bridge.process(); + ASSERT_EQ(symbols.size(), 1u); + + protocol::clear_file_atomic(layout.commands_log()); + for (std::uint64_t seq = 2; seq <= 6; ++seq) { + protocol::append_json_line( + layout.commands_log(), + protocol::make_file_jsonrpc_request( + seq, + "noop-" + std::to_string(seq), + "protocol.hello"), + config.max_line_bytes); + } + protocol::append_json_line( + layout.commands_log(), + protocol::make_file_jsonrpc_request( + 7, + "cmd-7", + "signal.submit", + make_signal_submit_params("bounded-regrow-7", "bounded-regrow-7", "GBPUSD")), + config.max_line_bytes); + + bridge.process(); + EXPECT_EQ(symbols.size(), 1u); + + bridge.process(); + EXPECT_EQ(symbols.size(), 1u); + + bridge.process(); + ASSERT_EQ(symbols.size(), 2u); + EXPECT_EQ(symbols.back(), "GBPUSD"); + + const auto checkpoint = protocol::read_json_file( + layout.commands_checkpoint(), + config.max_line_bytes); + EXPECT_EQ(checkpoint.at("last_file_seq").get(), 7u); +} + +TEST(MetaTraderFileBridge, DetectsRegrownLogWhenFirstCompleteLineIsMalformed) { + namespace mtfile = optionx::bridges::metatrader_file; + namespace protocol = optionx::bridges::metatrader_file::detail; + + const auto root = make_temp_root(); + ScopedPathCleanup cleanup(root); + + mtfile::MetaTraderFileBridgeConfig config; + config.common_files_root = root.u8string(); + config.bridge_id = 35; + config.client_id = "terminal-malformed-regrow"; + config.max_line_bytes = 8192; + config.max_scanned_records_per_poll = 2; + config.max_returned_records_per_poll = 1; + + const auto layout = protocol::make_layout(config); + protocol::ensure_runtime_directories(layout); + protocol::append_json_line( + layout.commands_log(), + protocol::make_file_jsonrpc_request( + 1, + "cmd-1", + "signal.submit", + make_signal_submit_params("malformed-regrow-1", "malformed-regrow-1")), + config.max_line_bytes); + + mtfile::MetaTraderFileBridge bridge; + ASSERT_TRUE(bridge.configure(std::make_unique(config))); + optionx::SignalId next_signal_id = 100; + bridge.on_signal_id() = [&]() { + return next_signal_id++; + }; + + std::vector symbols; + bridge.on_trade_signal() = [&](std::unique_ptr signal) { + ASSERT_TRUE(signal); + symbols.push_back(signal->symbol); + }; + + bridge.process(); + ASSERT_EQ(symbols.size(), 1u); + const auto old_size = std::filesystem::file_size(layout.commands_log()); + + protocol::clear_file_atomic(layout.commands_log()); + { + std::ofstream out(layout.commands_log(), std::ios::binary | std::ios::app); + ASSERT_TRUE(out); + out << nlohmann::json{{"broken", std::string(1024, 'x')}}.dump(-1) << '\n'; + } + protocol::append_json_line( + layout.commands_log(), + protocol::make_file_jsonrpc_request( + 2, + "cmd-2", + "signal.submit", + make_signal_submit_params("malformed-regrow-2", "malformed-regrow-2", "GBPUSD")), + config.max_line_bytes); + ASSERT_GE(std::filesystem::file_size(layout.commands_log()), old_size); + + bridge.process(); + ASSERT_EQ(symbols.size(), 2u); + EXPECT_EQ(symbols.back(), "GBPUSD"); +} + +TEST(MetaTraderFileBridge, RejectsTradeAffectingCommandWithoutIdempotencyKey) { + namespace mtfile = optionx::bridges::metatrader_file; + namespace protocol = optionx::bridges::metatrader_file::detail; + + const auto root = make_temp_root(); + ScopedPathCleanup cleanup(root); + + mtfile::MetaTraderFileBridgeConfig config; + config.common_files_root = root.u8string(); + config.bridge_id = 36; + config.client_id = "terminal-missing-key"; + config.max_line_bytes = 8192; + + const auto layout = protocol::make_layout(config); + protocol::ensure_runtime_directories(layout); + + protocol::append_json_line( + layout.commands_log(), + protocol::make_file_jsonrpc_request( + 1, + "cmd-1", + "signal.submit", + make_signal_submit_params("", "missing-key")), + config.max_line_bytes); + + mtfile::MetaTraderFileBridge bridge; + ASSERT_TRUE(bridge.configure(std::make_unique(config))); + bridge.on_signal_id() = []() { + return optionx::SignalId{1}; + }; + + int signal_count = 0; + bridge.on_trade_signal() = [&](std::unique_ptr) { + ++signal_count; + }; + + bridge.process(); + EXPECT_EQ(signal_count, 0); + + const auto events = protocol::read_ndjson_since_file_seq( + layout.events_log(), + 0, + config.max_line_bytes); + ASSERT_EQ(events.size(), 1u); + const auto& error = events[0].document.at("error"); + EXPECT_EQ(error.at("code").get(), -32602); +} + +TEST(MetaTraderFileBridge, RejectsTradeAffectingCommandWithoutValidUntil) { + namespace mtfile = optionx::bridges::metatrader_file; + namespace protocol = optionx::bridges::metatrader_file::detail; + + const auto root = make_temp_root(); + ScopedPathCleanup cleanup(root); + + mtfile::MetaTraderFileBridgeConfig config; + config.common_files_root = root.u8string(); + config.bridge_id = 46; + config.client_id = "terminal-missing-deadline"; + config.max_line_bytes = 8192; + + const auto layout = protocol::make_layout(config); + protocol::ensure_runtime_directories(layout); + + auto params = make_signal_submit_params("missing-deadline", "missing-deadline"); + params["context"].erase("valid_until_ms"); + protocol::append_json_line( + layout.commands_log(), + protocol::make_file_jsonrpc_request( + 1, + "cmd-1", + "signal.submit", + params), + config.max_line_bytes); + + mtfile::MetaTraderFileBridge bridge; + ASSERT_TRUE(bridge.configure(std::make_unique(config))); + bridge.on_signal_id() = []() { + return optionx::SignalId{1}; + }; + + int signal_count = 0; + bridge.on_trade_signal() = [&](std::unique_ptr) { + ++signal_count; + }; + + bridge.process(); + EXPECT_EQ(signal_count, 0); + + const auto events = protocol::read_ndjson_since_file_seq( + layout.events_log(), + 0, + config.max_line_bytes); + ASSERT_EQ(events.size(), 1u); + EXPECT_EQ(events[0].document.at("error").at("code").get(), -32602); +} + +TEST(MetaTraderFileBridge, RejectsSignalWhenTradeCallbackIsMissing) { + namespace mtfile = optionx::bridges::metatrader_file; + namespace protocol = optionx::bridges::metatrader_file::detail; + + const auto root = make_temp_root(); + ScopedPathCleanup cleanup(root); + + mtfile::MetaTraderFileBridgeConfig config; + config.common_files_root = root.u8string(); + config.bridge_id = 37; + config.client_id = "terminal-no-callback"; + config.max_line_bytes = 8192; + + const auto layout = protocol::make_layout(config); + protocol::ensure_runtime_directories(layout); + protocol::append_json_line( + layout.commands_log(), + protocol::make_file_jsonrpc_request( + 1, + "cmd-1", + "signal.submit", + make_signal_submit_params("missing-callback", "missing-callback")), + config.max_line_bytes); + + mtfile::MetaTraderFileBridge bridge; + ASSERT_TRUE(bridge.configure(std::make_unique(config))); + bridge.on_signal_id() = []() { + return optionx::SignalId{1}; + }; + + bridge.process(); + + const auto events = protocol::read_ndjson_since_file_seq( + layout.events_log(), + 0, + config.max_line_bytes); + ASSERT_EQ(events.size(), 1u); + const auto& result = events[0].document.at("result"); + EXPECT_EQ(result.at("status").get(), "rejected"); + EXPECT_EQ(result.at("reason").at("code").get(), "signal_handler_unavailable"); +} + +TEST(MetaTraderFileBridge, CorruptIdempotencyStateFailsClosed) { + namespace mtfile = optionx::bridges::metatrader_file; + namespace protocol = optionx::bridges::metatrader_file::detail; + + const auto root = make_temp_root(); + ScopedPathCleanup cleanup(root); + + mtfile::MetaTraderFileBridgeConfig config; + config.common_files_root = root.u8string(); + config.bridge_id = 47; + config.client_id = "terminal-corrupt-idempotency"; + config.max_line_bytes = 8192; + + const auto layout = protocol::make_layout(config); + protocol::ensure_runtime_directories(layout); + protocol::append_json_line( + layout.commands_log(), + protocol::make_file_jsonrpc_request( + 1, + "cmd-1", + "signal.submit", + make_signal_submit_params("corrupt-idempotency", "corrupt-idempotency")), + config.max_line_bytes); + protocol::write_json_file_atomic( + layout.idempotency_state(), + nlohmann::json{ + {"processed_through_file_seq", "not-a-number"}, + {"records", nlohmann::json::object()}, + {"tombstones", nlohmann::json::object()}, + {"request_index", nlohmann::json::object()} + }); + + mtfile::MetaTraderFileBridge bridge; + ASSERT_TRUE(bridge.configure(std::make_unique(config))); + bridge.on_signal_id() = []() { + return optionx::SignalId{1}; + }; + + int signal_count = 0; + bridge.on_trade_signal() = [&](std::unique_ptr) { + ++signal_count; + }; + + EXPECT_THROW(bridge.process(), std::runtime_error); + EXPECT_EQ(signal_count, 0); +} + +TEST(MetaTraderFileBridge, CorruptCommandCheckpointDoesNotReplayCommands) { + namespace mtfile = optionx::bridges::metatrader_file; + namespace protocol = optionx::bridges::metatrader_file::detail; + + const auto root = make_temp_root(); + ScopedPathCleanup cleanup(root); + + mtfile::MetaTraderFileBridgeConfig config; + config.common_files_root = root.u8string(); + config.bridge_id = 38; + config.client_id = "terminal-corrupt-checkpoint"; + config.max_line_bytes = 8192; + + const auto layout = protocol::make_layout(config); + protocol::ensure_runtime_directories(layout); + protocol::append_json_line( + layout.commands_log(), + protocol::make_file_jsonrpc_request( + 1, + "cmd-1", + "signal.submit", + make_signal_submit_params("corrupt-checkpoint", "corrupt-checkpoint")), + config.max_line_bytes); + protocol::write_json_file_atomic( + layout.commands_checkpoint(), + nlohmann::json{{"offset", 100}}); + + mtfile::MetaTraderFileBridge bridge; + ASSERT_TRUE(bridge.configure(std::make_unique(config))); + bridge.on_signal_id() = []() { + return optionx::SignalId{1}; + }; + + int signal_count = 0; + bridge.on_trade_signal() = [&](std::unique_ptr) { + ++signal_count; + }; + + EXPECT_THROW(bridge.process(), std::runtime_error); + EXPECT_EQ(signal_count, 0); +} + +TEST(MetaTraderFileBridge, DurableIdempotencySuppressesReplayAfterCheckpointLoss) { + namespace mtfile = optionx::bridges::metatrader_file; + namespace protocol = optionx::bridges::metatrader_file::detail; + + const auto root = make_temp_root(); + ScopedPathCleanup cleanup(root); + + mtfile::MetaTraderFileBridgeConfig config; + config.common_files_root = root.u8string(); + config.bridge_id = 39; + config.client_id = "terminal-idempotency"; + config.max_line_bytes = 8192; + + const auto layout = protocol::make_layout(config); + protocol::ensure_runtime_directories(layout); + protocol::append_json_line( + layout.commands_log(), + protocol::make_file_jsonrpc_request( + 1, + "cmd-1", + "signal.submit", + make_signal_submit_params("same-operation", "same-operation")), + config.max_line_bytes); + + { + mtfile::MetaTraderFileBridge bridge; + ASSERT_TRUE(bridge.configure(std::make_unique(config))); + bridge.on_signal_id() = []() { + return optionx::SignalId{10}; + }; + + int signal_count = 0; + bridge.on_trade_signal() = [&](std::unique_ptr) { + ++signal_count; + }; + bridge.process(); + EXPECT_EQ(signal_count, 1); + } + + std::filesystem::remove(layout.commands_checkpoint()); + + mtfile::MetaTraderFileBridge replay_bridge; + ASSERT_TRUE(replay_bridge.configure(std::make_unique(config))); + replay_bridge.on_signal_id() = []() { + return optionx::SignalId{11}; + }; + + int replay_count = 0; + replay_bridge.on_trade_signal() = [&](std::unique_ptr) { + ++replay_count; + }; + replay_bridge.process(); + EXPECT_EQ(replay_count, 0); + + const auto events = protocol::read_ndjson_since_file_seq( + layout.events_log(), + 0, + config.max_line_bytes); + ASSERT_EQ(events.size(), 1u); + EXPECT_EQ(events[0].document.at("result").at("signal_ref").at("signal_id").get(), "10"); +} + +TEST(MetaTraderFileBridge, DeduplicatesByJsonRpcId) { + namespace mtfile = optionx::bridges::metatrader_file; + namespace protocol = optionx::bridges::metatrader_file::detail; + + const auto root = make_temp_root(); + ScopedPathCleanup cleanup(root); + + mtfile::MetaTraderFileBridgeConfig config; + config.common_files_root = root.u8string(); + config.bridge_id = 40; + config.client_id = "terminal-rpc-id"; + config.max_line_bytes = 8192; + + const auto layout = protocol::make_layout(config); + protocol::ensure_runtime_directories(layout); + protocol::append_json_line( + layout.commands_log(), + protocol::make_file_jsonrpc_request( + 1, + "request-42", + "signal.submit", + make_signal_submit_params("key-A", "rpc-id-A")), + config.max_line_bytes); + protocol::append_json_line( + layout.commands_log(), + protocol::make_file_jsonrpc_request( + 2, + "request-42", + "signal.submit", + make_signal_submit_params("key-B", "rpc-id-B", "GBPUSD")), + config.max_line_bytes); + + mtfile::MetaTraderFileBridge bridge; + ASSERT_TRUE(bridge.configure(std::make_unique(config))); + bridge.on_signal_id() = []() { + return optionx::SignalId{50}; + }; + + int signal_count = 0; + bridge.on_trade_signal() = [&](std::unique_ptr) { + ++signal_count; + }; + bridge.process(); + EXPECT_EQ(signal_count, 1); + + const auto events = protocol::read_ndjson_since_file_seq( + layout.events_log(), + 0, + config.max_line_bytes); + ASSERT_EQ(events.size(), 2u); + EXPECT_EQ(events[0].document.at("result").at("signal_ref").at("signal_id").get(), "50"); + EXPECT_EQ(events[1].document.at("result").at("signal_ref").at("signal_id").get(), "50"); +} + +TEST(MetaTraderFileBridge, InDoubtIdempotencySuppressesReplayAfterCallbackFailure) { + namespace mtfile = optionx::bridges::metatrader_file; + namespace protocol = optionx::bridges::metatrader_file::detail; + + const auto root = make_temp_root(); + ScopedPathCleanup cleanup(root); + + mtfile::MetaTraderFileBridgeConfig config; + config.common_files_root = root.u8string(); + config.bridge_id = 41; + config.client_id = "terminal-in-doubt"; + config.max_line_bytes = 8192; + + const auto layout = protocol::make_layout(config); + protocol::ensure_runtime_directories(layout); + protocol::append_json_line( + layout.commands_log(), + protocol::make_file_jsonrpc_request( + 1, + "cmd-1", + "signal.submit", + make_signal_submit_params("in-doubt", "in-doubt")), + config.max_line_bytes); + + { + mtfile::MetaTraderFileBridge bridge; + ASSERT_TRUE(bridge.configure(std::make_unique(config))); + bridge.on_signal_id() = []() { + return optionx::SignalId{20}; + }; + + int signal_count = 0; + bridge.on_trade_signal() = [&](std::unique_ptr) { + ++signal_count; + throw std::runtime_error("simulated downstream crash window"); + }; + bridge.process(); + EXPECT_EQ(signal_count, 1); + } + + std::filesystem::remove(layout.commands_checkpoint()); + + mtfile::MetaTraderFileBridge replay_bridge; + ASSERT_TRUE(replay_bridge.configure(std::make_unique(config))); + replay_bridge.on_signal_id() = []() { + return optionx::SignalId{21}; + }; + + int replay_count = 0; + replay_bridge.on_trade_signal() = [&](std::unique_ptr) { + ++replay_count; + }; + replay_bridge.process(); + EXPECT_EQ(replay_count, 0); + + const auto events = protocol::read_ndjson_since_file_seq( + layout.events_log(), + 0, + config.max_line_bytes); + ASSERT_EQ(events.size(), 1u); + + const auto state = protocol::read_json_file( + layout.idempotency_state(), + config.max_idempotency_state_bytes); + ASSERT_TRUE(state.at("records").is_object()); + ASSERT_EQ(state.at("records").size(), 1u); + const auto& stored_result = + state.at("records").begin().value().at("result"); + EXPECT_EQ(stored_result.at("status").get(), "in_doubt"); + EXPECT_EQ(stored_result.at("signal_ref").at("signal_id").get(), "20"); +} + +TEST(MetaTraderFileBridge, PrunesIdempotencyRecords) { + namespace mtfile = optionx::bridges::metatrader_file; + namespace protocol = optionx::bridges::metatrader_file::detail; + + const auto root = make_temp_root(); + ScopedPathCleanup cleanup(root); + + mtfile::MetaTraderFileBridgeConfig config; + config.common_files_root = root.u8string(); + config.bridge_id = 42; + config.client_id = "terminal-idempotency-prune"; + config.max_line_bytes = 8192; + config.max_idempotency_records = 2; + config.max_idempotency_state_bytes = 8192; + + const auto layout = protocol::make_layout(config); + protocol::ensure_runtime_directories(layout); + const auto stable_deadline = protocol::unix_time_ms() + 60000; + for (std::uint64_t seq = 1; seq <= 3; ++seq) { + protocol::append_json_line( + layout.commands_log(), + protocol::make_file_jsonrpc_request( + seq, + "cmd-" + std::to_string(seq), + "signal.submit", + make_signal_submit_params( + "prune-" + std::to_string(seq), + "prune-" + std::to_string(seq), + "EURUSD", + "BUY", + stable_deadline)), + config.max_line_bytes); + } + + mtfile::MetaTraderFileBridge bridge; + ASSERT_TRUE(bridge.configure(std::make_unique(config))); + optionx::SignalId next_signal_id = 30; + bridge.on_signal_id() = [&]() { + return next_signal_id++; + }; + + int signal_count = 0; + bridge.on_trade_signal() = [&](std::unique_ptr) { + ++signal_count; + }; + bridge.process(); + EXPECT_EQ(signal_count, 3); + + const auto state = protocol::read_json_file( + layout.idempotency_state(), + config.max_idempotency_state_bytes); + ASSERT_TRUE(state.at("records").is_object()); + EXPECT_LE(state.at("records").size(), 2u); + EXPECT_EQ(state.at("processed_through_file_seq").get(), 3u); + + std::filesystem::remove(layout.commands_checkpoint()); + + mtfile::MetaTraderFileBridge replay_bridge; + ASSERT_TRUE(replay_bridge.configure(std::make_unique(config))); + replay_bridge.on_signal_id() = []() { + return optionx::SignalId{99}; + }; + + int replay_count = 0; + replay_bridge.on_trade_signal() = [&](std::unique_ptr) { + ++replay_count; + }; + replay_bridge.process(); + EXPECT_EQ(replay_count, 0); + + protocol::append_json_line( + layout.commands_log(), + protocol::make_file_jsonrpc_request( + 4, + "cmd-4", + "signal.submit", + make_signal_submit_params( + "prune-1", + "prune-1", + "EURUSD", + "BUY", + stable_deadline)), + config.max_line_bytes); + + replay_bridge.process(); + EXPECT_EQ(replay_count, 0); + + const auto after_reuse_events = protocol::read_ndjson_since_file_seq( + layout.events_log(), + 0, + config.max_line_bytes); + ASSERT_GE(after_reuse_events.size(), 4u); + EXPECT_EQ( + after_reuse_events.back().document.at("result").at("status").get(), + "accepted"); + EXPECT_EQ( + after_reuse_events.back().document.at("result").at("signal_ref").at("signal_id").get(), + "30"); +} + +TEST(MetaTraderFileBridge, ExpiresIdempotencyTombstonesAfterRetention) { + namespace mtfile = optionx::bridges::metatrader_file; + namespace protocol = optionx::bridges::metatrader_file::detail; + + const auto root = make_temp_root(); + ScopedPathCleanup cleanup(root); + + mtfile::MetaTraderFileBridgeConfig config; + config.common_files_root = root.u8string(); + config.bridge_id = 45; + config.client_id = "terminal-idempotency-expiry"; + config.max_line_bytes = 8192; + config.max_idempotency_records = 2; + config.max_idempotency_state_bytes = 8192; + config.idempotency_retention_ms = 24ull * 60ull * 60ull * 1000ull; + + const auto layout = protocol::make_layout(config); + protocol::ensure_runtime_directories(layout); + for (std::uint64_t seq = 1; seq <= 3; ++seq) { + protocol::append_json_line( + layout.commands_log(), + protocol::make_file_jsonrpc_request( + seq, + "cmd-" + std::to_string(seq), + "signal.submit", + make_signal_submit_params( + "expire-" + std::to_string(seq), + "expire-" + std::to_string(seq))), + config.max_line_bytes); + } + + mtfile::MetaTraderFileBridge bridge; + ASSERT_TRUE(bridge.configure(std::make_unique(config))); + optionx::SignalId next_signal_id = 70; + bridge.on_signal_id() = [&]() { + return next_signal_id++; + }; + + int signal_count = 0; + bridge.on_trade_signal() = [&](std::unique_ptr) { + ++signal_count; + }; + + bridge.process(); + EXPECT_EQ(signal_count, 3); + + auto state = protocol::read_json_file( + layout.idempotency_state(), + config.max_idempotency_state_bytes); + ASSERT_TRUE(state.at("tombstones").is_object()); + ASSERT_FALSE(state.at("tombstones").empty()); + for (auto it = state["tombstones"].begin(); it != state["tombstones"].end(); ++it) { + it.value()["evicted_at_ms"] = 1; + } + protocol::write_json_file_atomic(layout.idempotency_state(), state); + + protocol::append_json_line( + layout.commands_log(), + protocol::make_file_jsonrpc_request( + 4, + "cmd-4", + "signal.submit", + make_signal_submit_params("expire-1", "expire-1")), + config.max_line_bytes); + + mtfile::MetaTraderFileBridge replay_bridge; + ASSERT_TRUE(replay_bridge.configure(std::make_unique(config))); + replay_bridge.on_signal_id() = [&]() { + return next_signal_id++; + }; + + int replay_count = 0; + replay_bridge.on_trade_signal() = [&](std::unique_ptr) { + ++replay_count; + }; + + replay_bridge.process(); + EXPECT_EQ(replay_count, 1); + + const auto after = protocol::read_json_file( + layout.idempotency_state(), + config.max_idempotency_state_bytes); + ASSERT_TRUE(after.at("tombstones").is_object()); + for (const auto& item : after.at("tombstones").items()) { + EXPECT_NE(item.key(), "signal.submit\nexpire-1"); + } +} + +TEST(MetaTraderFileBridge, DoesNotPruneInDoubtRecords) { + namespace mtfile = optionx::bridges::metatrader_file; + namespace protocol = optionx::bridges::metatrader_file::detail; + + const auto root = make_temp_root(); + ScopedPathCleanup cleanup(root); + + mtfile::MetaTraderFileBridgeConfig config; + config.common_files_root = root.u8string(); + config.bridge_id = 43; + config.client_id = "terminal-in-doubt-retention"; + config.max_line_bytes = 8192; + config.max_idempotency_records = 1; + config.max_idempotency_state_bytes = 8192; + + const auto layout = protocol::make_layout(config); + protocol::ensure_runtime_directories(layout); + for (std::uint64_t seq = 1; seq <= 2; ++seq) { + protocol::append_json_line( + layout.commands_log(), + protocol::make_file_jsonrpc_request( + seq, + "cmd-" + std::to_string(seq), + "signal.submit", + make_signal_submit_params( + "in-doubt-retention-" + std::to_string(seq), + "in-doubt-retention-" + std::to_string(seq))), + config.max_line_bytes); + } + + mtfile::MetaTraderFileBridge bridge; + ASSERT_TRUE(bridge.configure(std::make_unique(config))); + optionx::SignalId next_signal_id = 60; + bridge.on_signal_id() = [&]() { + return next_signal_id++; + }; + + int signal_count = 0; + bridge.on_trade_signal() = [&](std::unique_ptr) { + ++signal_count; + throw std::runtime_error("keep operation in doubt"); + }; + + EXPECT_THROW(bridge.process(), std::runtime_error); + EXPECT_EQ(signal_count, 1); + + EXPECT_THROW(bridge.process(), std::runtime_error); + EXPECT_EQ(signal_count, 1); + + const auto state = protocol::read_json_file( + layout.idempotency_state(), + config.max_idempotency_state_bytes); + ASSERT_TRUE(state.at("records").is_object()); + ASSERT_EQ(state.at("records").size(), 1u); + EXPECT_EQ( + state.at("records").begin().value().at("result").at("status").get(), + "in_doubt"); +} + +TEST(MetaTraderFileBridge, AllowsReentrantTradeUpdatesFromSignalCallback) { + namespace mtfile = optionx::bridges::metatrader_file; + namespace protocol = optionx::bridges::metatrader_file::detail; + + const auto root = make_temp_root(); + ScopedPathCleanup cleanup(root); + + mtfile::MetaTraderFileBridgeConfig config; + config.common_files_root = root.u8string(); + config.bridge_id = 44; + config.client_id = "terminal-reentrant"; + config.max_line_bytes = 8192; + + const auto layout = protocol::make_layout(config); + protocol::ensure_runtime_directories(layout); + protocol::append_json_line( + layout.commands_log(), + protocol::make_file_jsonrpc_request( + 1, + "cmd-1", + "signal.submit", + make_signal_submit_params("reentrant", "reentrant")), + config.max_line_bytes); + + mtfile::MetaTraderFileBridge bridge; + ASSERT_TRUE(bridge.configure(std::make_unique(config))); + bridge.on_signal_id() = []() { + return optionx::SignalId{55}; + }; + + bridge.on_trade_signal() = [&](std::unique_ptr signal) { + ASSERT_TRUE(signal); + optionx::TradeRequest request; + request.trade_id = 55; + request.signal_id = signal->signal_id; + request.bridge_id = config.bridge_id; + request.symbol = signal->symbol; + request.order_type = signal->order_type; + request.option_type = signal->option_type; + request.amount = signal->amount; + + optionx::TradeResult result; + result.trade_id = 55; + result.trade_state = optionx::TradeState::OPEN_SUCCESS; + result.amount = signal->amount; + bridge.update_trade_result(request, result); + }; + + bridge.process(); + + const auto events = protocol::read_ndjson_since_file_seq( + layout.events_log(), + 0, + config.max_line_bytes); + ASSERT_EQ(events.size(), 2u); + EXPECT_EQ(events[0].document.at("method").get(), "trade.updated"); + EXPECT_EQ(events[1].document.at("result").at("status").get(), "accepted"); +} + +TEST(MetaTraderFileBridge, PublishesAccountQueriesAndTradeUpdates) { + namespace mtfile = optionx::bridges::metatrader_file; + namespace protocol = optionx::bridges::metatrader_file::detail; + + const auto root = make_temp_root(); + ScopedPathCleanup cleanup(root); + + mtfile::MetaTraderFileBridgeConfig config; + config.common_files_root = root.u8string(); + config.bridge_id = 32; + config.client_id = "terminal-02"; + config.max_line_bytes = 8192; + + const auto layout = protocol::make_layout(config); + protocol::ensure_runtime_directories(layout); + + mtfile::MetaTraderFileBridge bridge; + ASSERT_TRUE(bridge.configure(std::make_unique(config))); + + auto account = std::make_shared(); + account->user_id = 77; + account->balance = 2048.25; + account->currency = optionx::CurrencyType::EUR; + bridge.update_account_info(optionx::AccountInfoUpdate( + account, + optionx::AccountUpdateStatus::BALANCE_UPDATED)); + + protocol::append_json_line( + layout.commands_log(), + protocol::make_file_jsonrpc_request( + 1, + "balance-1", + "account.balance.get"), + config.max_line_bytes); + bridge.process(); + + optionx::TradeRequest request; + request.trade_id = 7; + request.signal_id = 12; + request.bridge_id = 32; + request.account_id = 77; + request.symbol = "BTCUSD"; + request.order_type = optionx::OrderType::SELL; + request.option_type = optionx::OptionType::SPRINT; + request.currency = optionx::CurrencyType::EUR; + request.amount = 3.0; + request.duration = 120; + + optionx::TradeResult trade_result; + trade_result.trade_id = 7; + trade_result.trade_state = optionx::TradeState::OPEN_SUCCESS; + trade_result.amount = 3.0; + trade_result.currency = optionx::CurrencyType::EUR; + trade_result.open_date = 12345; + + bridge.update_trade_result(request, trade_result); + + const auto state = protocol::read_json_file( + layout.state_snapshot(), + config.max_line_bytes); + ASSERT_EQ(state.at("accounts").size(), 1u); + EXPECT_EQ(state.at("accounts")[0].at("account_id").get(), "77"); + EXPECT_EQ(state.at("accounts")[0].at("balance").at("value").get(), "2048.25"); + EXPECT_EQ(state.at("accounts")[0].at("balance").at("currency").get(), "EUR"); + + const auto events = protocol::read_ndjson_since_file_seq( + layout.events_log(), + 0, + config.max_line_bytes); + ASSERT_EQ(events.size(), 3u); + EXPECT_EQ(events[0].document.at("method").get(), "balance.updated"); + EXPECT_EQ(events[1].document.at("id").get(), "balance-1"); + EXPECT_EQ(events[1].document.at("result").at("status").get(), "completed"); + EXPECT_EQ( + events[1].document.at("result").at("account").at("balance").at("value").get(), + "2048.25"); + EXPECT_EQ(events[2].document.at("method").get(), "trade.updated"); + const auto& trade = events[2].document.at("params").at("payload").at("trade"); + EXPECT_EQ(trade.at("trade_id").get(), "7"); + EXPECT_EQ(trade.at("account_id").get(), "77"); + EXPECT_EQ(trade.at("state").get(), "opened"); + EXPECT_FALSE(trade.at("final").get()); +} + int main(int argc, char** argv) { ::testing::InitGoogleTest(&argc, argv); return RUN_ALL_TESTS();