feat(telegram): add worker source adapter

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

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

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

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 <functional>
+#include <string>
+#include <utility>
+#include <vector>
+
+namespace optionx::bridges::telegram {
+
+    /// \struct TelegramWorkerSourceConfig
+    /// \brief Chat and topic selection for one worker live listener.
+    struct TelegramWorkerSourceConfig {
+        std::vector<std::string> chats;
+        std::vector<std::string> 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 <typename WorkerClient>
+    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

tests/telegram_worker_source_test.cpp

@@ -0,0 +1,104 @@
+#include <gtest/gtest.h>
+
+#include "optionx_cpp/bridges/telegram.hpp"
+
+#include <memory>
+#include <string>
+#include <vector>
+
+namespace {
+
+class FakeWorkerClient {
+public:
+    using handler_t = std::function<void(const nlohmann::json&)>;
+
+    bool start_listening(
+            const std::vector<std::string>& chats,
+            handler_t handler,
+            const std::vector<std::string>& 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<std::string> selected_chats;
+    std::vector<std::string> 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<optionx::bridges::telegram::TelegramRawMessage> messages;
+    std::vector<std::string> 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();
+}