From 2582d5eef004dde5478454a925417ab0007b18c5 Mon Sep 17 00:00:00 2001 From: Aster Seker Date: Sat, 1 Aug 2026 10:13:37 +0300 Subject: [PATCH 1/2] feat(telegram): add worker source adapter --- CMakeLists.txt | 1 + include/optionx_cpp/bridges/telegram.hpp | 1 + .../telegram/TelegramWorkerMessageSource.hpp | 137 ++++++++++++++++++ tests/telegram_worker_source_test.cpp | 104 +++++++++++++ 4 files changed, 243 insertions(+) create mode 100644 include/optionx_cpp/bridges/telegram/TelegramWorkerMessageSource.hpp create mode 100644 tests/telegram_worker_source_test.cpp diff --git a/CMakeLists.txt b/CMakeLists.txt index 2fa939f..c9e2b18 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -909,6 +909,7 @@ if(OPTIONX_BUILD_TESTS) telegram_dto_test telegram_signal_parser_test telegram_signal_bridge_test + telegram_worker_source_test ) if(OPTIONX_LIGHTWEIGHT_BRIDGE_SMOKE_TESTS) list(APPEND OPTIONX_LIGHTWEIGHT_TESTS diff --git a/include/optionx_cpp/bridges/telegram.hpp b/include/optionx_cpp/bridges/telegram.hpp index d00b914..962a744 100644 --- a/include/optionx_cpp/bridges/telegram.hpp +++ b/include/optionx_cpp/bridges/telegram.hpp @@ -10,5 +10,6 @@ #include "bridges/telegram/TelegramSignalParser.hpp" #include "bridges/telegram/TelegramSignalBridgeConfig.hpp" #include "bridges/telegram/TelegramSignalBridge.hpp" +#include "bridges/telegram/TelegramWorkerMessageSource.hpp" #endif // OPTIONX_HEADER_BRIDGES_TELEGRAM_HPP_INCLUDED diff --git a/include/optionx_cpp/bridges/telegram/TelegramWorkerMessageSource.hpp b/include/optionx_cpp/bridges/telegram/TelegramWorkerMessageSource.hpp new file mode 100644 index 0000000..c4eb5fa --- /dev/null +++ b/include/optionx_cpp/bridges/telegram/TelegramWorkerMessageSource.hpp @@ -0,0 +1,137 @@ +#pragma once +#ifndef OPTIONX_HEADER_BRIDGES_TELEGRAM_TELEGRAM_WORKER_MESSAGE_SOURCE_HPP_INCLUDED +#define OPTIONX_HEADER_BRIDGES_TELEGRAM_TELEGRAM_WORKER_MESSAGE_SOURCE_HPP_INCLUDED + +/// \file TelegramWorkerMessageSource.hpp +/// \brief Adapter from a tg-client-stdio-style worker client to Telegram bridge input. + +#include "bridges/telegram/TelegramSignalBridge.hpp" + +#include +#include +#include +#include + +namespace optionx::bridges::telegram { + + /// \struct TelegramWorkerSourceConfig + /// \brief Chat and topic selection for one worker live listener. + struct TelegramWorkerSourceConfig { + std::vector chats; + std::vector topic_ids; + }; + + /// \class TelegramWorkerMessageSource + /// \brief Binds a WorkerClient-like object to TelegramMessageSource. + /// + /// The worker type is a template so this header does not force OptionX to + /// include or link a particular worker repository. It is compatible with + /// `tg_client_stdio::WorkerClient` and deterministic test doubles that + /// provide the same `start_listening` and `stop_listening` operations. + template + class TelegramWorkerMessageSource final : public TelegramMessageSource { + public: + TelegramWorkerMessageSource( + WorkerClient& worker, + TelegramWorkerSourceConfig config) + : m_worker(worker), + m_config(std::move(config)) {} + + bool start(message_callback_t on_message, + error_callback_t on_error) override { + if (m_started || m_config.chats.empty() || !on_message) { + return false; + } + m_on_message = std::move(on_message); + m_on_error = std::move(on_error); + try { + if (!m_worker.start_listening( + m_config.chats, + [this](const auto& record) { handle_record(record); }, + m_config.topic_ids)) { + clear_callbacks(); + return false; + } + m_started = true; + return true; + } + catch (const std::exception& error) { + report_error(error.what()); + clear_callbacks(); + return false; + } + catch (...) { + report_error("Telegram worker listener failed to start."); + clear_callbacks(); + return false; + } + } + + void stop() noexcept override { + if (!m_started) { + clear_callbacks(); + return; + } + try { + (void)m_worker.stop_listening(); + } + catch (const std::exception& error) { + report_error(error.what()); + } + catch (...) { + report_error("Telegram worker listener failed to stop."); + } + m_started = false; + clear_callbacks(); + } + + private: + void handle_record(const nlohmann::json& record) { + if (record.value("message_type", "") == "error") { + const auto payload = record.value("payload", nlohmann::json::object()); + report_error(payload.value("message", "Telegram worker live error.")); + return; + } + if (record.value("operation", "") != "message.received") { + return; + } + try { + const auto payload = record.at("payload"); + const auto raw = TelegramRawMessage::from_json(payload.at("message")); + if (m_on_message) { + m_on_message(raw); + } + } + catch (const std::exception& error) { + report_error(error.what()); + } + catch (...) { + report_error("Telegram worker message record was invalid."); + } + } + + void report_error(const std::string& message) { + if (m_on_error) { + try { + m_on_error(message); + } + catch (...) { + } + } + } + + void clear_callbacks() noexcept { + m_on_message = {}; + m_on_error = {}; + } + + WorkerClient& m_worker; + TelegramWorkerSourceConfig m_config; + message_callback_t m_on_message; + error_callback_t m_on_error; + bool m_started = false; + }; + +} // namespace optionx::bridges::telegram + +#endif // OPTIONX_HEADER_BRIDGES_TELEGRAM_TELEGRAM_WORKER_MESSAGE_SOURCE_HPP_INCLUDED diff --git a/tests/telegram_worker_source_test.cpp b/tests/telegram_worker_source_test.cpp new file mode 100644 index 0000000..c422c5f --- /dev/null +++ b/tests/telegram_worker_source_test.cpp @@ -0,0 +1,104 @@ +#include + +#include "optionx_cpp/bridges/telegram.hpp" + +#include +#include +#include + +namespace { + +class FakeWorkerClient { +public: + using handler_t = std::function; + + bool start_listening( + const std::vector& chats, + handler_t handler, + const std::vector& topics) { + selected_chats = chats; + selected_topics = topics; + m_handler = std::move(handler); + return true; + } + + bool stop_listening() { + stopped = true; + m_handler = {}; + return true; + } + + void emit_message() { + m_handler(nlohmann::json{ + {"message_type", "event"}, + {"operation", "message.received"}, + {"request_id", 0}, + {"payload", { + {"message", { + {"chat_id", "-10042"}, + {"chat_title", "Signals"}, + {"topic_id", "7"}, + {"message_id", 12}, + {"date_ms", 1800000000000LL}, + {"text", "EURUSD BUY 5m"}, + {"media", nlohmann::json::array()}, + }} + }} + }); + } + + void emit_error() { + m_handler(nlohmann::json{ + {"message_type", "error"}, + {"request_id", 0}, + {"payload", {{"message", "listener failed"}}}, + }); + } + + std::vector selected_chats; + std::vector selected_topics; + bool stopped = false; + +private: + handler_t m_handler; +}; + +} // namespace + +TEST(TelegramWorkerMessageSource, AdaptsLiveRecordsAndErrors) { + FakeWorkerClient worker; + optionx::bridges::telegram::TelegramWorkerSourceConfig config; + config.chats = {"-10042"}; + config.topic_ids = {"7"}; + optionx::bridges::telegram::TelegramWorkerMessageSource source(worker, config); + + std::vector messages; + std::vector errors; + ASSERT_TRUE(source.start( + [&](const auto& message) { messages.push_back(message); }, + [&](const auto& error) { errors.push_back(error); })); + worker.emit_message(); + worker.emit_error(); + + ASSERT_EQ(messages.size(), 1u); + EXPECT_EQ(messages.front().message_identity(), "telegram:-10042:7:12"); + ASSERT_EQ(errors.size(), 1u); + EXPECT_EQ(errors.front(), "listener failed"); + EXPECT_EQ(worker.selected_chats, config.chats); + EXPECT_EQ(worker.selected_topics, config.topic_ids); + + source.stop(); + EXPECT_TRUE(worker.stopped); +} + +TEST(TelegramWorkerMessageSource, RejectsEmptyChatSelection) { + FakeWorkerClient worker; + optionx::bridges::telegram::TelegramWorkerMessageSource source( + worker, {}); + EXPECT_FALSE(source.start([](const auto&) {}, [](const auto&) {})); +} + +int main(int argc, char** argv) { + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} From eddb6a5b50668035bc12beba9975817d409942ed Mon Sep 17 00:00:00 2001 From: Aster Seker Date: Sat, 1 Aug 2026 10:14:14 +0300 Subject: [PATCH 2/2] docs(telegram): update worker adapter status --- guides/telegram-bridge-design.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/guides/telegram-bridge-design.md b/guides/telegram-bridge-design.md index 3886050..b040b8e 100644 --- a/guides/telegram-bridge-design.md +++ b/guides/telegram-bridge-design.md @@ -335,7 +335,7 @@ Completed without an authorized Telegram session: Next steps: 1. Merge and pin the worker repository's supervisor/archive PRs. -2. Add `TelegramMessageSource` adapter code around `WorkerClient` and test it +2. Pin the merged worker repository in an OptionX consumer and run the adapter against the mock worker process. 3. Add a historical archive/parser fixture example. 4. Perform the first real authorization, proxy and live-channel check with an