feat(gateway): full Telegram+Discord integration with agent loop
Complete multi-platform gateway wiring: config: - [platform.telegram] section with bot_token - [platform.discord] section with bot_token + guild_id - Threaded through GyandootConfig → AgentConfig via typed structs gateway: - GatewayRunner::new(platforms) takes Vec<Box<dyn Platform>> - start() spawns all platform pollers, merges receivers - serve(agent) runs full message loop: receive → access control → typing indicator → session_turn() → platform.send() - Ctrl+C graceful shutdown via tokio::select! cmd_serve now: 1. Loads config, creates agent 2. Builds Telegram + Discord adapters from config 3. Starts gateway pollers 4. Enters serve loop until Ctrl+C
This commit is contained in:
@@ -9,6 +9,8 @@ description = "GyanDoot v2 — Gateway for platform sessions, routing, streaming
|
||||
gyandoot-core = { path = "../gyandoot-core" }
|
||||
gyandoot-memory = { path = "../gyandoot-memory" }
|
||||
gyandoot-providers = { path = "../gyandoot-providers" }
|
||||
gyandoot-platform = { path = "../gyandoot-platform" }
|
||||
gyandoot-agent = { path = "../gyandoot-agent" }
|
||||
tokio.workspace = true
|
||||
serde.workspace = true
|
||||
serde_json.workspace = true
|
||||
|
||||
@@ -2,14 +2,26 @@
|
||||
//!
|
||||
//! Gateway for platform sessions, message routing, and streaming delivery.
|
||||
//! Connects platform adapters (Telegram, Discord, Webhook) to the agent core.
|
||||
//!
|
||||
//! ## Architecture
|
||||
//!
|
||||
//! ```text
|
||||
//! Telegram ──┐
|
||||
//! Discord ──┤──→ mpsc::Receiver<IncomingMessage> ──→ AgentLoop ──→ response → platform.send()
|
||||
//! Webhook ──┘
|
||||
//! ```
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::Mutex;
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use tokio::sync::{Mutex, mpsc};
|
||||
use uuid::Uuid;
|
||||
use chrono::Utc;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use tracing::info;
|
||||
use tracing::{error, info, warn};
|
||||
|
||||
use gyandoot_agent::agent_loop::AgentLoop;
|
||||
use gyandoot_platform::{IncomingMessage, OutgoingMessage, Platform};
|
||||
|
||||
/// Unique session identifier.
|
||||
pub type SessionId = Uuid;
|
||||
@@ -97,7 +109,7 @@ impl AccessControl {
|
||||
pub fn check_user(&self, username: &str) -> bool {
|
||||
match &self.allowed_users {
|
||||
Some(users) => users.iter().any(|u| u.eq_ignore_ascii_case(username)),
|
||||
None => true, // empty = allow all
|
||||
None => true,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -115,19 +127,31 @@ impl Default for AccessControl {
|
||||
}
|
||||
}
|
||||
|
||||
/// Multi-platform adapter list.
|
||||
type PlatformList = Arc<Mutex<Vec<Box<dyn Platform>>>>;
|
||||
|
||||
/// The main gateway runner — orchestrates platform connections and message flow.
|
||||
pub struct GatewayRunner {
|
||||
pub sessions: Arc<Mutex<HashMap<SessionId, GatewaySession>>>,
|
||||
pub access_control: AccessControl,
|
||||
running: Arc<Mutex<bool>>,
|
||||
running: Arc<AtomicBool>,
|
||||
platforms: PlatformList,
|
||||
/// Receivers from all platform pollers (merged into one).
|
||||
rx: Option<mpsc::Receiver<IncomingMessage>>,
|
||||
}
|
||||
|
||||
impl GatewayRunner {
|
||||
pub fn new() -> Self {
|
||||
/// Create a new gateway with platforms.
|
||||
///
|
||||
/// Takes ownership of platform adapters. At least one must be provided
|
||||
/// (though you can add more later with `add_platform`).
|
||||
pub fn new(platforms: Vec<Box<dyn Platform>>) -> Self {
|
||||
Self {
|
||||
sessions: Arc::new(Mutex::new(HashMap::new())),
|
||||
access_control: AccessControl::new(),
|
||||
running: Arc::new(Mutex::new(false)),
|
||||
running: Arc::new(AtomicBool::new(false)),
|
||||
platforms: Arc::new(Mutex::new(platforms)),
|
||||
rx: None,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -136,21 +160,185 @@ impl GatewayRunner {
|
||||
self
|
||||
}
|
||||
|
||||
pub async fn start(&self) -> anyhow::Result<()> {
|
||||
let mut running = self.running.lock().await;
|
||||
*running = true;
|
||||
info!("Gateway started");
|
||||
/// Add a platform adapter after construction.
|
||||
pub async fn add_platform(&self, platform: Box<dyn Platform>) {
|
||||
let mut platforms = self.platforms.lock().await;
|
||||
platforms.push(platform);
|
||||
info!("Platform adapter added (total: {})", platforms.len());
|
||||
}
|
||||
|
||||
/// Start all platform adapters and begin routing messages.
|
||||
///
|
||||
/// This spawns polling loops for each platform and merges their message
|
||||
/// receivers. Does NOT block — call `serve(agent)` to process messages.
|
||||
pub async fn start(&mut self) -> anyhow::Result<()> {
|
||||
self.running.store(true, Ordering::SeqCst);
|
||||
|
||||
let platforms = self.platforms.lock().await;
|
||||
if platforms.is_empty() {
|
||||
info!("Gateway started with no platform adapters");
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
// Start each platform and collect receivers
|
||||
let mut all_rxs = Vec::new();
|
||||
for platform in platforms.iter() {
|
||||
match platform.start().await {
|
||||
Ok(rx) => {
|
||||
info!("Platform '{}' started successfully", platform.name());
|
||||
all_rxs.push((platform.name().to_string(), rx));
|
||||
}
|
||||
Err(e) => {
|
||||
warn!("Platform '{}' failed to start: {}", platform.name(), e);
|
||||
// Non-fatal — other platforms may still work
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if all_rxs.is_empty() {
|
||||
warn!("No platform adapters started successfully");
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
// Merge all receivers into one via a select loop
|
||||
let (tx, rx) = mpsc::channel::<IncomingMessage>(1024);
|
||||
self.rx = Some(rx);
|
||||
|
||||
let running = Arc::clone(&self.running);
|
||||
tokio::spawn(async move {
|
||||
loop {
|
||||
if !running.load(Ordering::SeqCst) {
|
||||
break;
|
||||
}
|
||||
|
||||
// Check all platform receivers
|
||||
for (name, rx) in &mut all_rxs {
|
||||
match rx.try_recv() {
|
||||
Ok(msg) => {
|
||||
if tx.send(msg).await.is_err() {
|
||||
warn!("Gateway message channel closed");
|
||||
return;
|
||||
}
|
||||
}
|
||||
Err(mpsc::error::TryRecvError::Disconnected) => {
|
||||
warn!("Platform '{name}' receiver disconnected");
|
||||
}
|
||||
Err(mpsc::error::TryRecvError::Empty) => {}
|
||||
}
|
||||
}
|
||||
|
||||
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
|
||||
}
|
||||
});
|
||||
|
||||
info!("Gateway started with {} platform(s)", platforms.len());
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn stop(&self) {
|
||||
let mut running = self.running.lock().await;
|
||||
*running = false;
|
||||
/// Run the gateway message processing loop with an agent.
|
||||
///
|
||||
/// Blocks until the gateway is stopped, forwarding incoming messages
|
||||
/// to the agent and routing responses back through the correct platform.
|
||||
pub async fn serve(&mut self, agent: &AgentLoop) -> anyhow::Result<()> {
|
||||
// Ensure the gateway is running
|
||||
if !self.running.load(Ordering::SeqCst) {
|
||||
self.start().await?;
|
||||
}
|
||||
|
||||
let mut rx = self.rx.take().ok_or_else(|| {
|
||||
anyhow::anyhow!("Gateway not started — call start() first or use serve() directly")
|
||||
})?;
|
||||
|
||||
info!("Gateway serve loop started — listening for messages");
|
||||
|
||||
// Init agent session
|
||||
agent.init_session().await.map_err(|e| {
|
||||
anyhow::anyhow!("Agent session init failed: {}", e)
|
||||
})?;
|
||||
|
||||
let platforms = Arc::clone(&self.platforms);
|
||||
let sessions = Arc::clone(&self.sessions);
|
||||
|
||||
while self.running.load(Ordering::SeqCst) {
|
||||
tokio::select! {
|
||||
Some(msg) = rx.recv() => {
|
||||
// Access control
|
||||
if !self.access_control.check_user(&msg.username) {
|
||||
warn!("Access denied for user: {}", msg.username);
|
||||
continue;
|
||||
}
|
||||
if !self.access_control.check_channel(&msg.channel_id) {
|
||||
warn!("Access denied for channel: {}", msg.channel_id);
|
||||
continue;
|
||||
}
|
||||
|
||||
// Create/update session
|
||||
{
|
||||
let source = SessionSource {
|
||||
platform: msg.platform.clone(),
|
||||
chat_id: msg.channel_id.clone(),
|
||||
thread_id: None,
|
||||
};
|
||||
let mut sessions = sessions.lock().await;
|
||||
// Find existing session for this source or create new
|
||||
let existing = sessions.values_mut().find(|s| {
|
||||
s.source.platform == msg.platform
|
||||
&& s.source.chat_id == msg.channel_id
|
||||
});
|
||||
match existing {
|
||||
Some(session) => {
|
||||
session.last_active = Utc::now();
|
||||
session.message_count += 1;
|
||||
}
|
||||
None => {
|
||||
let session = GatewaySession::new(source);
|
||||
sessions.insert(session.id, session);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Send typing indicator
|
||||
let platforms_guard = platforms.lock().await;
|
||||
if let Some(p) = platforms_guard.iter().find(|p| p.name() == msg.platform) {
|
||||
let _ = p.send_typing(&msg.channel_id).await;
|
||||
}
|
||||
|
||||
// Process with agent
|
||||
let response = match agent.session_turn(&msg.text).await {
|
||||
Ok(text) => text,
|
||||
Err(e) => {
|
||||
error!("Agent error: {}", e);
|
||||
format!("⚠️ Error processing your message: {}", e)
|
||||
}
|
||||
};
|
||||
|
||||
// Send response back through the same platform
|
||||
if let Some(p) = platforms_guard.iter().find(|p| p.name() == msg.platform) {
|
||||
let outgoing = OutgoingMessage {
|
||||
channel_id: msg.channel_id.clone(),
|
||||
text: response,
|
||||
reply_to: Some(msg.message_id),
|
||||
};
|
||||
if let Err(e) = p.send(outgoing).await {
|
||||
error!("Failed to send response on {}: {}", msg.platform, e);
|
||||
}
|
||||
}
|
||||
}
|
||||
else => break,
|
||||
}
|
||||
}
|
||||
|
||||
info!("Gateway serve loop finished");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn stop(&self) {
|
||||
self.running.store(false, Ordering::SeqCst);
|
||||
info!("Gateway stopped");
|
||||
}
|
||||
|
||||
pub async fn is_running(&self) -> bool {
|
||||
*self.running.lock().await
|
||||
pub fn is_running(&self) -> bool {
|
||||
self.running.load(Ordering::SeqCst)
|
||||
}
|
||||
|
||||
pub async fn create_session(&self, source: SessionSource) -> SessionId {
|
||||
@@ -167,12 +355,6 @@ impl GatewayRunner {
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for GatewayRunner {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
@@ -210,11 +392,11 @@ mod tests {
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_gateway_lifecycle() {
|
||||
let gw = GatewayRunner::new();
|
||||
assert!(!gw.is_running().await);
|
||||
let mut gw = GatewayRunner::new(vec![]);
|
||||
assert!(!gw.is_running());
|
||||
gw.start().await.unwrap();
|
||||
assert!(gw.is_running().await);
|
||||
gw.stop().await;
|
||||
assert!(!gw.is_running().await);
|
||||
assert!(gw.is_running());
|
||||
gw.stop();
|
||||
assert!(!gw.is_running());
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user