From fb109a9ae40c71e0e9fbcd9ae8df09db8e34c542 Mon Sep 17 00:00:00 2001 From: buddhism5080 Date: Thu, 13 Aug 2026 11:37:00 +0800 Subject: [PATCH] fix(telegram): use real forum topic id in unified create_thread UnifiedGatewayAdapter::create_thread previously returned the triggering message_id as thread_id after fire-and-forget create_topic. Telegram forum replies then used a message id as message_thread_id, causing "message thread not found" while a real topic had already been created. Await create_topic GatewayResponse (oneshot via AppState.pending_commands), demux responses on the unified event bridge before process_gateway_event, and return the platform message_thread_id. Also stop mis-parsing GatewayResponse as GatewayEvent (null-as-string warnings). Add GH Actions workflow to build --features unified arm64/x64 artifacts for install without local compile. --- .github/workflows/build-unified-binary.yml | 60 ++++++++++++ Cargo.toml | 1 + crates/openab-gateway/src/lib.rs | 13 +++ src/main.rs | 18 +++- src/unified_adapter.rs | 107 ++++++++++++++++++--- 5 files changed, 183 insertions(+), 16 deletions(-) create mode 100644 .github/workflows/build-unified-binary.yml diff --git a/.github/workflows/build-unified-binary.yml b/.github/workflows/build-unified-binary.yml new file mode 100644 index 000000000..66b04cd3f --- /dev/null +++ b/.github/workflows/build-unified-binary.yml @@ -0,0 +1,60 @@ +# Build a unified openab binary for local install (e.g. arm64 host) without +# compiling on the developer machine. Triggered on this branch and manually. + +name: Build Unified Binary (PR) + +on: + push: + branches: + - fix/telegram-unified-create-thread + workflow_dispatch: + pull_request: + paths: + - "src/**" + - "crates/**" + - "Cargo.toml" + - "Cargo.lock" + - ".github/workflows/build-unified-binary.yml" + +jobs: + build-unified: + strategy: + fail-fast: false + matrix: + include: + - { target: aarch64-unknown-linux-gnu, runner: ubuntu-24.04-arm, os: linux, arch: arm64 } + - { target: x86_64-unknown-linux-gnu, runner: ubuntu-latest, os: linux, arch: x64 } + runs-on: ${{ matrix.runner }} + permissions: + contents: read + steps: + - uses: actions/checkout@v6 + + - name: Install Rust + uses: dtolnay/rust-toolchain@stable + with: + targets: ${{ matrix.target }} + + - uses: Swatinem/rust-cache@v2 + with: + key: unified-${{ matrix.target }} + + - name: Build unified release binary + run: cargo build --release --features unified --target ${{ matrix.target }} + + - name: Package + shell: bash + run: | + NAME="openab-unified-${{ matrix.os }}-${{ matrix.arch }}" + mkdir -p dist + cp "target/${{ matrix.target }}/release/openab" "dist/openab-unified" + cp config.toml.example dist/ 2>/dev/null || true + tar -C dist -czf "${NAME}.tar.gz" . + ls -la "${NAME}.tar.gz" + + - name: Upload artifact + uses: actions/upload-artifact@v4 + with: + name: openab-unified-${{ matrix.os }}-${{ matrix.arch }} + path: openab-unified-*.tar.gz + retention-days: 14 diff --git a/Cargo.toml b/Cargo.toml index 5a834a13f..e54daf99b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -25,6 +25,7 @@ serde = { version = "1", features = ["derive"] } serde_json = "1" async-trait = "0.1" serenity = { version = "0.12", default-features = false, features = ["client", "gateway", "model", "rustls_backend", "cache"], optional = true } +uuid = { version = "1", features = ["v4"] } [features] # Default: core only (Discord + Slack). Gateway ships as separate binary. diff --git a/crates/openab-gateway/src/lib.rs b/crates/openab-gateway/src/lib.rs index d2b257c7d..2881384cf 100644 --- a/crates/openab-gateway/src/lib.rs +++ b/crates/openab-gateway/src/lib.rs @@ -95,6 +95,16 @@ pub struct AppState { /// Optional pre-download identity probe (see [`IngressTrustProbe`]). pub trust_probe: Option, pub client: reqwest::Client, + /// In-process oneshot waiters for gateway command responses + /// (`create_topic`, etc.). Used by the unified binary so `create_thread` + /// can await the real Telegram `message_thread_id` instead of inventing + /// one. The standalone gateway WS path keeps its own `pending` map on + /// `GatewayAdapter` and does not use this field. + pub pending_commands: Arc< + tokio::sync::Mutex< + HashMap>, + >, + >, } @@ -145,6 +155,7 @@ impl AppState { lineworks_ingress_queue: Arc::new(Semaphore::new(LINEWORKS_INGRESS_QUEUE_MAX)), trust_probe: None, client: reqwest::Client::new(), + pending_commands: Arc::new(tokio::sync::Mutex::new(HashMap::new())), } } @@ -267,6 +278,7 @@ impl AppState { lineworks_ingress_queue: Arc::new(Semaphore::new(LINEWORKS_INGRESS_QUEUE_MAX)), trust_probe: None, client, + pending_commands: Arc::new(tokio::sync::Mutex::new(HashMap::new())), } } @@ -849,6 +861,7 @@ pub async fn serve(config: ServeConfig) -> anyhow::Result<()> { lineworks_ingress_queue: Arc::new(Semaphore::new(LINEWORKS_INGRESS_QUEUE_MAX)), trust_probe: None, client, + pending_commands: Arc::new(tokio::sync::Mutex::new(HashMap::new())), }); // Phase 1 L1 audit (#1356): warn if any active webhook platform has no diff --git a/src/main.rs b/src/main.rs index a2ee786ac..acce81bf3 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1481,9 +1481,12 @@ async fn main() -> anyhow::Result<()> { let app = app.with_state(gw_state.clone()); // Bridge task: receive events from adapters via event_tx, dispatch to core - let unified_adapter: Arc = Arc::new( - unified_adapter::UnifiedGatewayAdapter::new(gw_state.clone()), - ); + // Keep a typed Arc so create_topic responses can complete oneshot waiters + // on UnifiedGatewayAdapter before ordinary GatewayEvent processing. + let unified_adapter_impl = Arc::new(unified_adapter::UnifiedGatewayAdapter::new( + gw_state.clone(), + )); + let unified_adapter: Arc = unified_adapter_impl.clone(); // Bot gating still reads env here (structural, not L2/L3): // channel/user gating moved to the shared trust registry, seeded @@ -1515,15 +1518,22 @@ async fn main() -> anyhow::Result<()> { filestore: filestore.clone(), }); - // Spawn the event bridge (event_tx → process_gateway_event) + // Spawn the event bridge (event_tx → process_gateway_event). + // Demux GatewayResponse first so create_topic waiters receive the + // real message_thread_id and responses are not mis-parsed as events. let mut event_rx = event_tx.subscribe(); let bridge_ctx = event_ctx.clone(); + let bridge_adapter = unified_adapter_impl.clone(); tokio::spawn(async move { loop { match event_rx.recv().await { Ok(event_json) => { let ctx = bridge_ctx.clone(); + let adapter = bridge_adapter.clone(); tokio::spawn(async move { + if adapter.try_complete_pending(&event_json).await { + return; + } if let Err(e) = process_gateway_event(&event_json, &ctx).await { tracing::warn!(error = %e, "unified bridge: event processing failed"); } diff --git a/src/unified_adapter.rs b/src/unified_adapter.rs index c943f4d0d..7df95d6b3 100644 --- a/src/unified_adapter.rs +++ b/src/unified_adapter.rs @@ -1,14 +1,19 @@ //! UnifiedGatewayAdapter — routes ChatAdapter calls through in-process gateway //! platform adapters based on the ChannelRef.platform field. -use anyhow::Result; +use anyhow::{anyhow, Result}; use async_trait::async_trait; use openab_core::adapter::{ChannelRef, ChatAdapter, MessageRef}; -use openab_gateway::schema::{Content, GatewayReply, ReplyChannel}; +use openab_gateway::schema::{Content, GatewayReply, GatewayResponse, ReplyChannel}; use openab_gateway::AppState; use std::collections::HashMap; use std::sync::Arc; use tokio::sync::Mutex; +use tracing::warn; +use uuid::Uuid; + +/// How long `create_thread` waits for the platform `create_topic` response. +const CREATE_TOPIC_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5); pub struct UnifiedGatewayAdapter { pub gw_state: Arc, @@ -24,6 +29,34 @@ impl UnifiedGatewayAdapter { } } + /// If `json` is a gateway command response (`openab.gateway.response.v1`), + /// deliver it to any waiter registered via [`Self::create_thread`] and + /// return `true`. Returns `false` for ordinary inbound events so the + /// caller can keep processing them as `GatewayEvent`s. + /// + /// Without this demux, `createForumTopic` responses are fed into + /// `process_gateway_event`, which fails with + /// `invalid type: null, expected a string` (response fields do not match + /// the event schema) and the real `message_thread_id` is discarded. + pub async fn try_complete_pending(&self, json: &str) -> bool { + let Ok(resp) = serde_json::from_str::(json) else { + return false; + }; + if resp.schema != "openab.gateway.response.v1" { + return false; + } + if let Some(tx) = self + .gw_state + .pending_commands + .lock() + .await + .remove(&resp.request_id) + { + let _ = tx.send(resp); + } + true + } + /// Dispatch a GatewayReply to the correct platform adapter. async fn dispatch_reply(&self, reply: &GatewayReply) { let client = &self.gw_state.client; @@ -168,19 +201,69 @@ impl ChatAdapter for UnifiedGatewayAdapter { async fn create_thread( &self, channel: &ChannelRef, - trigger_msg: &MessageRef, + _trigger_msg: &MessageRef, title: &str, ) -> Result { - let reply = self.build_reply(channel, title, Some("create_topic"), None); + // Mirror the standalone GatewayAdapter path: register a oneshot, send + // create_topic with request_id, wait for GatewayResponse carrying the + // real platform thread id (Telegram message_thread_id). + // + // The previous implementation ignored the response and returned + // trigger_msg.message_id as thread_id. For Telegram forums that value + // is a *message* id, not a topic id, so replies fail with + // "Bad Request: message thread not found" after a topic was created. + let req_id = format!("req_{}", Uuid::new_v4()); + let (tx, rx) = tokio::sync::oneshot::channel(); + self.gw_state + .pending_commands + .lock() + .await + .insert(req_id.clone(), tx); + + let mut reply = self.build_reply(channel, title, Some("create_topic"), None); + reply.request_id = Some(req_id.clone()); self.dispatch_reply(&reply).await; - // Return a thread channel ref with the trigger message as thread_id - Ok(ChannelRef { - platform: channel.platform.clone(), - channel_id: channel.channel_id.clone(), - thread_id: Some(trigger_msg.message_id.clone()), - parent_id: Some(channel.channel_id.clone()), - origin_event_id: channel.origin_event_id.clone(), - }) + + match tokio::time::timeout(CREATE_TOPIC_TIMEOUT, rx).await { + Ok(Ok(resp)) if resp.success => { + if let Some(thread_id) = resp.thread_id { + Ok(ChannelRef { + platform: channel.platform.clone(), + channel_id: channel.channel_id.clone(), + thread_id: Some(thread_id), + parent_id: Some(channel.channel_id.clone()), + origin_event_id: channel.origin_event_id.clone(), + }) + } else { + warn!( + request_id = %req_id, + "create_topic succeeded but thread_id missing; falling back to parent channel" + ); + Ok(channel.clone()) + } + } + Ok(Ok(resp)) => { + warn!( + err = ?resp.error, + request_id = %req_id, + "create_topic failed, falling back to same channel" + ); + Ok(channel.clone()) + } + Ok(Err(_)) => { + // oneshot dropped — treat as failure + self.gw_state.pending_commands.lock().await.remove(&req_id); + Err(anyhow!("create_topic response channel closed")) + } + Err(_) => { + warn!( + request_id = %req_id, + "create_topic timeout, falling back to same channel" + ); + self.gw_state.pending_commands.lock().await.remove(&req_id); + Ok(channel.clone()) + } + } } async fn add_reaction(&self, msg: &MessageRef, emoji: &str) -> Result<()> {