Add QUIC operator transport

CMakeLists.txt

@@ -70,6 +70,15 @@ IF (FLEET_PROTOCOL_BUILD_EXTERNAL_SERVER)
     FIND_PACKAGE(Boost CONFIG REQUIRED)
     FIND_PACKAGE(ZLIB REQUIRED)
     FIND_PACKAGE(cpprestsdk REQUIRED)
+
+    # Operator-facing QUIC transport (BAF-1744), opt-in alongside the Fleet HTTP API above —
+    # ported from teleop-module's equivalent (BAF-1670): msquic + protobuf for the operator
+    # stream wire protocol (own proto/transparent_operator_stream.proto — see its file header
+    # for why it can't share teleop-module's operator_stream.proto).
+    FIND_PACKAGE(msquic CONFIG REQUIRED)
+    FIND_PACKAGE(Protobuf REQUIRED)
+    PROTOBUF_GENERATE_CPP(TRANSPARENT_OPERATOR_PROTO_SRCS TRANSPARENT_OPERATOR_PROTO_HDRS
+            proto/transparent_operator_stream.proto)
 ENDIF ()
 
 CMDEF_ADD_LIBRARY(
@@ -117,8 +126,13 @@ IF (FLEET_PROTOCOL_BUILD_EXTERNAL_SERVER)
             LIBRARY_GROUP transparent-external-server
             TYPE SHARED
             SOURCES "source/external_server_api.cpp" "source/memory_management.cpp" "source/device_management.cpp"
+            "source/transparent_operator_stream/OperatorChannel.cpp"
+            "source/transparent_operator_stream/QuicOperatorServer.cpp"
+            ${TRANSPARENT_OPERATOR_PROTO_SRCS}
             VERSION ${TRANSPARENT_MODULE_VERSION}
     )
+    # Generated transparent_operator_stream.pb.h lands in the binary dir.
+    TARGET_INCLUDE_DIRECTORIES(transparent-external-server-shared PRIVATE ${CMAKE_CURRENT_BINARY_DIR})
     # Explicitly add Boost dependencies due to obsolete cpprestsdk cmake export package
     TARGET_LINK_LIBRARIES(fleet-http-client-shared::fleet-http-client-shared
             INTERFACE
@@ -130,6 +144,8 @@ IF (FLEET_PROTOCOL_BUILD_EXTERNAL_SERVER)
             PRIVATE
             transparent_module_sources
             fleet-http-client-shared::fleet-http-client-shared
+            msquic
+            protobuf::libprotobuf
             PUBLIC
             fleet-protocol-interface::module-maintainer-external-server-interface
     )

cmake/Dependencies.cmake

@@ -11,5 +11,10 @@ IF (FLEET_PROTOCOL_BUILD_EXTERNAL_SERVER)
     BA_PACKAGE_LIBRARY(boost         v1.86.0)
     BA_PACKAGE_LIBRARY(cpprestsdk    v2.10.20)
     BA_PACKAGE_LIBRARY(zlib                                 v1.3.2)
+    # Operator-facing QUIC transport (BAF-1744, ported from teleop-module's BAF-1670): msquic +
+    # protobuf for the operator stream protocol (own .proto — see its file header for why it can't
+    # share teleop-module's).
+    BA_PACKAGE_LIBRARY(msquic                                v2.5.6)
+    BA_PACKAGE_LIBRARY(protobuf                              v4.21.12)
 
 ENDIF ()

include/bringauto/transparent_module_utils/external_server_api_structures.hpp

@@ -2,8 +2,11 @@
 
 #include <bringauto/fleet_protocol/cxx/DeviceID.hpp>
 #include <bringauto/fleet_protocol/http_client/FleetApiClient.hpp>
+#include <bringauto/transparent_module_utils/operator_stream/OperatorChannel.hpp>
+#include <bringauto/transparent_module_utils/operator_stream/QuicOperatorServer.hpp>
 
 #include <condition_variable>
+#include <memory>
 #include <string>
 #include <thread>
 #include <vector>
@@ -19,6 +22,12 @@ namespace bringauto::transparent_module_utils
         std::mutex mutex;
         std::condition_variable con_variable;
         long last_command_timestamp;
+
+        /// Operator-facing QUIC transport (BAF-1744) — used instead of the Fleet HTTP API above
+        /// when the config supplies quic_port; otherwise both stay unused/idle. Declared in this
+        /// order so operator_channel outlives quic_server (the server holds a reference to it).
+        operator_stream::OperatorChannel operator_channel;
+        std::unique_ptr<operator_stream::QuicOperatorServer> quic_server;
     };
 
 } // namespace bringauto::transparent_module_utils

include/bringauto/transparent_module_utils/operator_stream/OperatorChannel.hpp

@@ -0,0 +1,60 @@
+#pragma once
+
+#include <chrono>
+#include <condition_variable>
+#include <cstdint>
+#include <mutex>
+#include <optional>
+#include <string>
+#include <vector>
+
+namespace bringauto::transparent_module_utils::operator_stream {
+
+/**
+ * One command received from an operator over QUIC, handed across the thread boundary to the
+ * synchronous ES API. `payload` is the opaque command buffer (today: the BAF-1651 JSON envelope)
+ * passed through verbatim; device_* are the fleet-protocol device coordinates kept for routing.
+ *
+ * Ported from teleop-module's OperatorChannel (BAF-1670) for the Transparent module's QUIC
+ * operator transport (BAF-1744) — see that repo's equivalent file for the original design notes.
+ */
+struct OperatorCommand {
+	std::vector<std::uint8_t> payload;
+	std::int64_t timestamp_ms{0};
+	std::uint32_t module_id{0};
+	std::uint32_t device_type{0};
+	std::string device_role;
+	std::string device_name;
+};
+
+/**
+ * Thread-safe seam between the QUIC server's ASYNCHRONOUS msquic receive callback (producer) and
+ * the module's SYNCHRONOUS, blocking external-server API (consumer) — see
+ * QuicOperatorServer for the producer side and external_server_api.cpp's wait_for_command() for
+ * the consumer side.
+ */
+class OperatorChannel {
+public:
+	/// Producer (QUIC recv-callback thread): hand over a command and wake a waiter. Coalesces to
+	/// the newest — a pending unconsumed command is dropped (freshest wins). No-op after shutdown().
+	void push(OperatorCommand command);
+
+	/// Consumer (ES `wait_for_command` thread): block up to `timeout` for the latest command.
+	/// Returns std::nullopt on timeout or after shutdown().
+	[[nodiscard]] std::optional<OperatorCommand> waitForCommand(std::chrono::milliseconds timeout);
+
+	/// Drop queued commands — e.g. on operator disconnect.
+	void clear();
+
+	/// Unblock all waiters and refuse further waits/pushes (module teardown).
+	void shutdown();
+
+private:
+	std::mutex mutex_;
+	std::condition_variable cv_;
+	/// Single coalescing slot: push() always supersedes any unconsumed command, so at most one is held.
+	std::optional<OperatorCommand> pending_;
+	bool shutdown_{false};
+};
+
+} // namespace bringauto::transparent_module_utils::operator_stream

include/bringauto/transparent_module_utils/operator_stream/QuicOperatorServer.hpp

@@ -0,0 +1,132 @@
+#pragma once
+
+#include <bringauto/transparent_module_utils/operator_stream/OperatorChannel.hpp>
+
+#include <msquic.h>
+
+#include <atomic>
+#include <cstdint>
+#include <mutex>
+#include <string>
+
+namespace bringauto::transparent_module_utils::operator_stream {
+
+/**
+ * Configuration for the operator-facing QUIC server. mTLS optional (disable client auth +
+ * self-signed server cert for loopback bring-up).
+ */
+struct QuicOperatorServerConfig {
+	std::uint16_t port{0};
+	std::string alpn{"transparent-operator"};
+	std::string certPath;          // server certificate (PEM)
+	std::string keyPath;           // server private key (PEM)
+	std::string caCertsPath;       // CA for verifying operator client certs (empty = no CA file)
+	bool disableClientAuth{false}; // true for loopback/dev; false requires operator client certs
+	std::string listenAddress;     // empty = any
+};
+
+/**
+ * Operator-facing QUIC server living inside the Transparent module's external-server plugin —
+ * the operator leg, used alongside (not instead of) the existing Fleet HTTP API path (BAF-1744;
+ * ported from teleop-module's QuicOperatorServer, BAF-1670, single-operator Phase 1).
+ *
+ * Wire format is protobuf (proto/transparent_operator_stream.proto's OperatorMessage), matching
+ * teleop-module's QuicOperatorServer this was ported from — but its own .proto/package, since
+ * external-server-cpp dlopens both teleop-external-server-shared.so and
+ * transparent-external-server-shared.so into the same process, and protobuf's generated-descriptor
+ * registry is process-global: two .so's registering the same .proto file/package would crash. The
+ * BAF-1651 JSON envelope itself still rides as opaque `bytes` inside OperatorMessage's
+ * StatusUpdate/CommandRequest — this class stays decoupled from that schema.
+ *
+ *   cloud -> operator : sendStatus() opens a short unidirectional stream per OperatorMessage{status}
+ *                       (START|FIN), freeing the send buffer on SEND_COMPLETE.
+ *   operator -> cloud : each operator command arrives as a unidirectional stream; the receive
+ *                       callback accumulates the bytes, parses OperatorMessage{command}, and hands
+ *                       it to the OperatorChannel (which unblocks the module's wait_for_command()).
+ *
+ * Threading: msquic invokes the callbacks on its own worker thread(s). Commands cross into the
+ * synchronous module API through the OperatorChannel (the thread-safe seam). sendStatus() is called
+ * on the ES thread (from forward_status) and msquic StreamSend is thread-safe.
+ */
+class QuicOperatorServer {
+public:
+	QuicOperatorServer(QuicOperatorServerConfig config, OperatorChannel &channel);
+	~QuicOperatorServer();
+
+	QuicOperatorServer(const QuicOperatorServer &) = delete;
+	QuicOperatorServer &operator=(const QuicOperatorServer &) = delete;
+
+	/// Open msquic, registration, configuration, credential (mTLS), and the listener. False on any failure.
+	[[nodiscard]] bool initialize();
+
+	/// Start the listener. False on failure.
+	[[nodiscard]] bool start();
+
+	/// Stop the listener + drop the operator connection.
+	void stop();
+
+	/// Outcome of sendStatus(). NoOperator is an expected best-effort drop (nobody to send to),
+	/// distinct from SendFailed (serialize / StreamOpen / StreamSend error) so the caller can react.
+	enum class SendResult { Sent, NoOperator, SendFailed };
+
+	/// Send one status to the connected operator (best-effort: NoOperator if none). Called from the
+	/// ES thread. payload is the opaque status buffer (today: JSON); device_* are the fleet-protocol coords.
+	SendResult sendStatus(std::uint32_t module_id, std::uint32_t device_type, const std::string &device_role,
+						  const std::string &device_name, const std::uint8_t *payload, std::size_t payload_size,
+						  std::int64_t timestamp_ms);
+
+	[[nodiscard]] bool hasOperator() const { return operatorConnection_.load() != nullptr; }
+
+private:
+	/// Heap buffer kept alive across an async StreamSend; freed in the send callback on SEND_COMPLETE.
+	struct SendBuffer {
+		explicit SendBuffer(std::string &&data) : payload(std::move(data)) {
+			buffer.Length = static_cast<std::uint32_t>(payload.size());
+			buffer.Buffer = reinterpret_cast<std::uint8_t *>(payload.data());
+		}
+		std::string payload;
+		QUIC_BUFFER buffer{};
+	};
+
+	/// Per-inbound-stream accumulation context (a command may arrive in multiple RECEIVE events).
+	struct InboundStreamContext {
+		QuicOperatorServer *server{nullptr};
+		HQUIC connection{nullptr}; // owning connection; commands are accepted only while it is the operator
+		std::string bytes;
+	};
+
+	bool loadQuic();
+	bool initRegistration();
+	bool initConfiguration();
+	bool initCredential();
+	bool initListener();
+
+	/// Parse an accumulated command frame and hand it to the channel.
+	void handleCommandBytes(const std::string &bytes);
+
+	static QUIC_STATUS QUIC_API listenerCallback(HQUIC listener, void *context, QUIC_LISTENER_EVENT *event);
+	static QUIC_STATUS QUIC_API connectionCallback(HQUIC connection, void *context, QUIC_CONNECTION_EVENT *event);
+	static QUIC_STATUS QUIC_API commandStreamCallback(HQUIC stream, void *context, QUIC_STREAM_EVENT *event);
+	static QUIC_STATUS QUIC_API sendStreamCallback(HQUIC stream, void *context, QUIC_STREAM_EVENT *event);
+
+	QuicOperatorServerConfig config_;
+	OperatorChannel &channel_;
+
+	const QUIC_API_TABLE *quic_{nullptr};
+	HQUIC registration_{nullptr};
+	HQUIC quicConfig_{nullptr};
+	HQUIC listener_{nullptr};
+	QUIC_ADDR quicAddr_{};
+	QUIC_BUFFER alpnBuffer_{};
+
+	/// Single active operator connection (Phase 1: single operator).
+	std::atomic<HQUIC> operatorConnection_{nullptr};
+	std::atomic<bool> running_{false};
+
+	/// Serializes use of the connection handle (sendStatus' StreamOpen/StreamSend) against its
+	/// teardown (ConnectionClose in the SHUTDOWN_COMPLETE callback). The atomic above guards the
+	/// pointer value only, not the lifetime of the handle it points to.
+	std::mutex connectionMutex_;
+};
+
+} // namespace bringauto::transparent_module_utils::operator_stream

proto/transparent_operator_stream.proto

@@ -0,0 +1,58 @@
+syntax = "proto3";
+
+// GUI <-> transparent-module QUIC stream protocol (BAF-1744), modeled on teleop-module's
+// operator_stream.proto (BAF-1670) — same message shapes, but its own package/file: external-server-cpp
+// dlopens both teleop-external-server-shared.so and transparent-external-server-shared.so into the
+// same process, and protobuf's generated-descriptor registry is process-global, so two .so's
+// registering the same .proto file/package would crash. A distinct package avoids that collision.
+//
+// Transport: each message travels on its own UNIDIRECTIONAL QUIC stream between the GUI (client)
+// and transparent-module's external-server plugin (server). The sender opens a stream, writes the
+// serialized OperatorMessage, and closes it with FIN — the stream boundary frames the message, so
+// there is NO length prefix and one stream carries exactly one OperatorMessage.
+//
+// Direction:
+//   rtsp-server -> GUI : StatusUpdate   (device-reported status / BAF-1651 response envelope)
+//   GUI -> rtsp-server  : CommandRequest (a BAF-1651 command envelope)
+//
+// The BAF-1651 JSON envelope ({type, version, action, id, timestamp, payload}) is passed through
+// VERBATIM as opaque bytes — this protocol only carries the Fleet Protocol device_id + timestamp
+// needed to route the message, and stays decoupled from the BAF-1651 schema.
+// NB: `operator_stream`, not `operator` — `operator` is a C++ keyword (invalid generated namespace).
+package bringauto.transparent.operator_stream;
+
+// Fleet-protocol device coordinates of the addressed device (module 3 / type 1 / role rtsp_server).
+message DeviceId {
+    uint32 module_id = 1;
+    uint32 type      = 2;
+    string role      = 3;
+    string name      = 4;
+}
+
+// rtsp-server -> GUI: one status/response push.
+message StatusUpdate {
+    DeviceId device       = 1;
+    bytes    payload      = 2;   // opaque BAF-1651 JSON envelope
+    int64    timestamp_ms = 3;
+}
+
+// GUI -> rtsp-server: one streaming-control command.
+message CommandRequest {
+    DeviceId device       = 1;
+    bytes    payload      = 2;   // opaque BAF-1651 JSON envelope
+    int64    timestamp_ms = 3;
+}
+
+// GUI -> rtsp-server: sent once on connect. Single-operator (Phase 1 of the pattern this was
+// ported from) — contents ignored for now.
+message OperatorHello {
+    string operator_id = 1;
+}
+
+message OperatorMessage {
+    oneof kind {
+        OperatorHello  hello   = 1;   // GUI -> rtsp-server
+        StatusUpdate   status  = 2;   // rtsp-server -> GUI
+        CommandRequest command = 3;   // GUI -> rtsp-server
+    }
+}