mirror of
https://github.com/tinyhumansai/openhuman.git
synced 2026-07-27 21:08:00 +00:00
feat(event_bus): wire webhooks, channels & skills through the event bus (#379)
* feat(event_bus): enhance domain event handling across modules - Added new `DomainEvent` variants for channel and skill events, including `ChannelMessageReceived`, `ChannelMessageProcessed`, `ChannelConnected`, `ChannelDisconnected`, `SkillLoaded`, `SkillStopped`, and `SkillStartFailed`. - Implemented event publishing in the channels and skills modules to track message processing and skill lifecycle events. - Created dedicated event bus handler files for the skills and webhooks domains, preparing for future subscriber implementations. - Updated documentation in `CLAUDE.md` to reflect the new domain events and their usage. These changes improve the observability and modularity of the system by leveraging an event-driven architecture for cross-module communication. * feat(event_bus): implement channel and webhook event handling - Introduced `ChannelInboundSubscriber` to handle inbound channel messages, triggering the agent inference loop and sending replies via the backend REST API. - Added `WebhookRequestSubscriber` to manage incoming webhook requests, routing them to the appropriate skill and handling responses. - Updated the global event bus initialization in `bootstrap_skill_runtime` to register both channel and webhook subscribers. - Enhanced `DomainEvent` with new variants for channel inbound messages and webhook requests, improving event-driven communication across modules. These changes enhance the modularity and responsiveness of the system by leveraging an event-driven architecture for channel and webhook interactions. * refactor(event_bus): update domain event documentation and subscriber initialization - Revised the documentation in `CLAUDE.md` to provide a concise overview of domain events and their associated subscriber files, enhancing clarity for future development. - Updated the `start_channels` function to initialize `WebhookRequestSubscriber` and `ChannelInboundSubscriber`, ensuring proper event handling for webhooks and channel messages. - Streamlined the event bus subscriber registration process, reinforcing the modular architecture of the system. These changes improve the maintainability and usability of the event bus framework, facilitating better cross-module communication. * style: apply cargo fmt formatting Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * fix(event_bus): remove duplicate subscriber registration in start_channels WebhookRequestSubscriber and ChannelInboundSubscriber were registered in both bootstrap_skill_runtime() and start_channels(), causing events to be handled twice when both paths run in the same process. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * fix(event_bus): prevent subscriber handles from being dropped on function exit SubscriptionHandle::drop aborts the background task. Since bootstrap_skill_runtime() returns immediately after setup, the local handles were dropped, cancelling both subscribers. Use std::mem::forget to leak the handles so the tasks live for the entire process. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * fix(event_bus): ensure subscriber handles persist beyond function exit Modified the handling of subscriber registration to prevent premature dropping of handles in `bootstrap_skill_runtime()`. This change ensures that the background tasks for subscribers remain active for the entire process lifecycle, enhancing event handling reliability. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * fix(webhooks): use proper JSON serialization for error response bodies Hand-escaped JSON strings only handled double quotes, not backslashes, newlines, or other control chars. Replaced with serde_json serialization via an error_body() helper. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * fix(webhooks): use proper JSON serialization for error response bodies Hand-escaped JSON strings only handled double quotes, not backslashes, newlines, or other control chars. Replaced with serde_json serialization via an error_body() helper. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Steven Enamakel
co-authored by
Claude Opus 4.6
Claude Opus 4.6
parent
c5acfa90e0
commit
07b1df4f24
@@ -311,6 +311,14 @@ A typed pub/sub event bus for **decoupled cross-module communication**. The bus
|
||||
| `event_bus::subscribe_global(handler)` | Subscribe from anywhere (returns `None` if not yet initialized) |
|
||||
| `event_bus::global()` | Get `Option<&'static EventBus>` for advanced use |
|
||||
|
||||
**Domains:** `agent`, `memory`, `channel`, `cron`, `skill`, `tool`, `webhook`, `system`. See `events.rs` for the full variant list — events carry rich payloads so subscribers have everything they need.
|
||||
|
||||
**Domain subscriber files** — each domain owns its `bus.rs` with `EventHandler` impls:
|
||||
- `cron/bus.rs` — `CronDeliverySubscriber` (delivers job output to channels)
|
||||
- `webhooks/bus.rs` — `WebhookRequestSubscriber` (routes incoming requests to skills, emits responses via socket)
|
||||
- `channels/bus.rs` — `ChannelInboundSubscriber` (runs agent loop for inbound socket messages)
|
||||
- `skills/bus.rs` — stub for future subscribers
|
||||
|
||||
**Adding events for a new domain:**
|
||||
|
||||
1. Add variants to `DomainEvent` in `events.rs` (prefix with domain name, e.g. `BillingInvoiceCreated { ... }`).
|
||||
|
||||
@@ -689,6 +689,22 @@ pub async fn bootstrap_skill_runtime() {
|
||||
let _ = std::fs::create_dir_all(&workspace_dir);
|
||||
engine.set_workspace_dir(workspace_dir);
|
||||
|
||||
// --- Event bus bootstrap ---
|
||||
// Ensure the global event bus is initialized (no-op if already done by start_channels).
|
||||
let bus =
|
||||
crate::openhuman::event_bus::init_global(crate::openhuman::event_bus::DEFAULT_CAPACITY);
|
||||
// Register domain subscribers for cross-module event handling.
|
||||
// Leak the handles so the background tasks live for the entire process —
|
||||
// SubscriptionHandle::drop aborts the task, and bootstrap_skill_runtime()
|
||||
// returns immediately after setup.
|
||||
std::mem::forget(bus.subscribe(Arc::new(
|
||||
crate::openhuman::webhooks::bus::WebhookRequestSubscriber::new(),
|
||||
)));
|
||||
std::mem::forget(bus.subscribe(Arc::new(
|
||||
crate::openhuman::channels::bus::ChannelInboundSubscriber::new(),
|
||||
)));
|
||||
log::info!("[event_bus] webhook and channel subscribers registered");
|
||||
|
||||
// --- Socket manager bootstrap ---
|
||||
let socket_mgr = Arc::new(SocketManager::new());
|
||||
set_global_socket_manager(socket_mgr.clone());
|
||||
|
||||
@@ -0,0 +1,179 @@
|
||||
//! Event bus handlers for the channels domain.
|
||||
//!
|
||||
//! The [`ChannelInboundSubscriber`] handles inbound channel messages published
|
||||
//! by the socket transport layer. It runs the agent inference loop via the web
|
||||
//! channel provider and sends the reply back through the REST API.
|
||||
|
||||
use crate::openhuman::event_bus::{DomainEvent, EventHandler};
|
||||
use async_trait::async_trait;
|
||||
use serde_json::json;
|
||||
|
||||
/// Subscribes to `ChannelInboundMessage` events and runs the agent loop,
|
||||
/// sending replies back to the originating channel via the backend REST API.
|
||||
pub struct ChannelInboundSubscriber;
|
||||
|
||||
impl ChannelInboundSubscriber {
|
||||
pub fn new() -> Self {
|
||||
Self
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl EventHandler for ChannelInboundSubscriber {
|
||||
fn name(&self) -> &str {
|
||||
"channel::inbound_handler"
|
||||
}
|
||||
|
||||
fn domains(&self) -> Option<&[&str]> {
|
||||
Some(&["channel"])
|
||||
}
|
||||
|
||||
async fn handle(&self, event: &DomainEvent) {
|
||||
let DomainEvent::ChannelInboundMessage {
|
||||
event_name: _,
|
||||
channel,
|
||||
message,
|
||||
raw_data: _,
|
||||
} = event
|
||||
else {
|
||||
return;
|
||||
};
|
||||
|
||||
tracing::info!(
|
||||
"[channel-inbound] received message from channel='{}' len={}",
|
||||
channel,
|
||||
message.len()
|
||||
);
|
||||
|
||||
let thread_id = format!("channel:{}", channel);
|
||||
let client_id = "inbound".to_string();
|
||||
|
||||
let mut event_rx =
|
||||
crate::openhuman::channels::providers::web::subscribe_web_channel_events();
|
||||
|
||||
let request_id = match crate::openhuman::channels::providers::web::start_chat(
|
||||
&client_id, &thread_id, message, None, None,
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(rid) => {
|
||||
tracing::debug!(
|
||||
"[channel-inbound] agent started request_id={} thread={}",
|
||||
rid,
|
||||
thread_id
|
||||
);
|
||||
rid
|
||||
}
|
||||
Err(err) => {
|
||||
tracing::error!("[channel-inbound] start_chat failed: {}", err);
|
||||
send_channel_reply(
|
||||
channel,
|
||||
&format!("Sorry, I couldn't process your message: {err}"),
|
||||
)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
let timeout = tokio::time::Duration::from_secs(180);
|
||||
let deadline = tokio::time::Instant::now() + timeout;
|
||||
|
||||
loop {
|
||||
tokio::select! {
|
||||
event = event_rx.recv() => {
|
||||
match event {
|
||||
Ok(ev) if ev.request_id == request_id => {
|
||||
if ev.event == "chat_done" || ev.event == "chat:done" {
|
||||
let reply = ev.full_response.unwrap_or_default();
|
||||
if reply.trim().is_empty() {
|
||||
tracing::warn!("[channel-inbound] agent returned empty response");
|
||||
return;
|
||||
}
|
||||
tracing::info!(
|
||||
"[channel-inbound] agent done, replying to channel='{}' len={}",
|
||||
channel,
|
||||
reply.len()
|
||||
);
|
||||
send_channel_reply(channel, &reply).await;
|
||||
return;
|
||||
}
|
||||
if ev.event == "chat_error" || ev.event == "chat:error" {
|
||||
let err_msg = ev.message.unwrap_or_else(|| "unknown error".to_string());
|
||||
tracing::error!("[channel-inbound] agent error: {}", err_msg);
|
||||
send_channel_reply(
|
||||
channel,
|
||||
&format!("Sorry, I encountered an error: {err_msg}"),
|
||||
)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
}
|
||||
Ok(_) => {}
|
||||
Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
|
||||
tracing::warn!("[channel-inbound] event bus lagged, skipped {} events", n);
|
||||
}
|
||||
Err(tokio::sync::broadcast::error::RecvError::Closed) => {
|
||||
tracing::error!("[channel-inbound] event bus closed unexpectedly");
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
_ = tokio::time::sleep_until(deadline) => {
|
||||
tracing::error!("[channel-inbound] agent timed out after {}s", timeout.as_secs());
|
||||
send_channel_reply(channel, "Sorry, the request timed out.").await;
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Send a text reply back to a channel via the backend REST API.
|
||||
async fn send_channel_reply(channel: &str, text: &str) {
|
||||
let config = match crate::openhuman::config::rpc::load_config_with_timeout().await {
|
||||
Ok(c) => c,
|
||||
Err(e) => {
|
||||
tracing::error!("[channel-inbound] failed to load config: {}", e);
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
let api_url = crate::api::config::effective_api_url(&config.api_url);
|
||||
let jwt = match crate::api::jwt::get_session_token(&config) {
|
||||
Ok(Some(t)) => t,
|
||||
Ok(None) => {
|
||||
tracing::error!("[channel-inbound] no session JWT — cannot reply");
|
||||
return;
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::error!("[channel-inbound] failed to get session token: {}", e);
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
let client = match crate::api::rest::BackendOAuthClient::new(&api_url) {
|
||||
Ok(c) => c,
|
||||
Err(e) => {
|
||||
tracing::error!("[channel-inbound] failed to create API client: {}", e);
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
let body = json!({ "text": text });
|
||||
match client.send_channel_message(channel, &jwt, body).await {
|
||||
Ok(resp) => {
|
||||
tracing::info!(
|
||||
"[channel-inbound] reply sent to channel='{}' response={:?}",
|
||||
channel,
|
||||
resp
|
||||
);
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::error!(
|
||||
"[channel-inbound] failed to send reply to channel='{}': {}",
|
||||
channel,
|
||||
e
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,5 +1,6 @@
|
||||
//! Channel implementations and runtime orchestration.
|
||||
|
||||
pub mod bus;
|
||||
pub mod cli;
|
||||
pub mod controllers;
|
||||
pub mod providers;
|
||||
|
||||
@@ -11,6 +11,7 @@ use crate::openhuman::channels::routes::{
|
||||
};
|
||||
use crate::openhuman::channels::traits;
|
||||
use crate::openhuman::channels::{Channel, SendMessage};
|
||||
use crate::openhuman::event_bus::{publish_global, DomainEvent};
|
||||
use crate::openhuman::providers::{self, ChatMessage};
|
||||
use crate::openhuman::util::truncate_with_ellipsis;
|
||||
use std::sync::Arc;
|
||||
@@ -73,6 +74,13 @@ pub(crate) async fn process_channel_message(
|
||||
truncate_with_ellipsis(&msg.content, 80)
|
||||
);
|
||||
|
||||
publish_global(DomainEvent::ChannelMessageReceived {
|
||||
channel: msg.channel.clone(),
|
||||
sender: msg.sender.clone(),
|
||||
content: msg.content.clone(),
|
||||
thread_ts: msg.thread_ts.clone(),
|
||||
});
|
||||
|
||||
let target_channel = ctx.channels_by_name.get(&msg.channel).cloned();
|
||||
if handle_runtime_command_if_needed(ctx.as_ref(), &msg, target_channel.as_ref()).await {
|
||||
return;
|
||||
@@ -243,7 +251,7 @@ pub(crate) async fn process_channel_message(
|
||||
log_worker_join_result(handle.await);
|
||||
}
|
||||
|
||||
match llm_result {
|
||||
let (success, response_text) = match llm_result {
|
||||
Ok(Ok(response)) => {
|
||||
// Save user + assistant turn to per-sender history
|
||||
{
|
||||
@@ -280,7 +288,7 @@ pub(crate) async fn process_channel_message(
|
||||
}
|
||||
} else if let Err(e) = channel
|
||||
.send(
|
||||
&SendMessage::new(response, &msg.reply_target)
|
||||
&SendMessage::new(&response, &msg.reply_target)
|
||||
.in_thread(msg.thread_ts.clone()),
|
||||
)
|
||||
.await
|
||||
@@ -288,6 +296,7 @@ pub(crate) async fn process_channel_message(
|
||||
eprintln!(" ❌ Failed to reply on {}: {e}", channel.name());
|
||||
}
|
||||
}
|
||||
(true, response)
|
||||
}
|
||||
Ok(Err(e)) => {
|
||||
if is_context_window_overflow_error(&e) {
|
||||
@@ -316,9 +325,19 @@ pub(crate) async fn process_channel_message(
|
||||
.await;
|
||||
}
|
||||
}
|
||||
|
||||
publish_global(DomainEvent::ChannelMessageProcessed {
|
||||
channel: msg.channel.clone(),
|
||||
sender: msg.sender.clone(),
|
||||
content: msg.content.clone(),
|
||||
response: error_text.to_string(),
|
||||
elapsed_ms: started_at.elapsed().as_millis() as u64,
|
||||
success: false,
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
let error_response = format!("⚠️ Error: {e}");
|
||||
eprintln!(
|
||||
" ❌ LLM error after {}ms: {e}",
|
||||
started_at.elapsed().as_millis()
|
||||
@@ -326,17 +345,18 @@ pub(crate) async fn process_channel_message(
|
||||
if let Some(channel) = target_channel.as_ref() {
|
||||
if let Some(ref draft_id) = draft_message_id {
|
||||
let _ = channel
|
||||
.finalize_draft(&msg.reply_target, draft_id, &format!("⚠️ Error: {e}"))
|
||||
.finalize_draft(&msg.reply_target, draft_id, &error_response)
|
||||
.await;
|
||||
} else {
|
||||
let _ = channel
|
||||
.send(
|
||||
&SendMessage::new(format!("⚠️ Error: {e}"), &msg.reply_target)
|
||||
&SendMessage::new(&error_response, &msg.reply_target)
|
||||
.in_thread(msg.thread_ts.clone()),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
}
|
||||
(false, error_response)
|
||||
}
|
||||
Err(_) => {
|
||||
let timeout_msg = format!("LLM response timed out after {}s", ctx.message_timeout_secs);
|
||||
@@ -345,24 +365,34 @@ pub(crate) async fn process_channel_message(
|
||||
timeout_msg,
|
||||
started_at.elapsed().as_millis()
|
||||
);
|
||||
let error_text =
|
||||
"⚠️ Request timed out while waiting for the model. Please try again.".to_string();
|
||||
if let Some(channel) = target_channel.as_ref() {
|
||||
let error_text =
|
||||
"⚠️ Request timed out while waiting for the model. Please try again.";
|
||||
if let Some(ref draft_id) = draft_message_id {
|
||||
let _ = channel
|
||||
.finalize_draft(&msg.reply_target, draft_id, error_text)
|
||||
.finalize_draft(&msg.reply_target, draft_id, &error_text)
|
||||
.await;
|
||||
} else {
|
||||
let _ = channel
|
||||
.send(
|
||||
&SendMessage::new(error_text, &msg.reply_target)
|
||||
&SendMessage::new(&error_text, &msg.reply_target)
|
||||
.in_thread(msg.thread_ts.clone()),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
}
|
||||
(false, error_text)
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
publish_global(DomainEvent::ChannelMessageProcessed {
|
||||
channel: msg.channel.clone(),
|
||||
sender: msg.sender.clone(),
|
||||
content: msg.content.clone(),
|
||||
response: response_text,
|
||||
elapsed_ms: started_at.elapsed().as_millis() as u64,
|
||||
success,
|
||||
});
|
||||
}
|
||||
|
||||
pub(crate) async fn run_message_dispatch_loop(
|
||||
|
||||
@@ -45,6 +45,9 @@ pub async fn start_channels(config: Config) -> Result<()> {
|
||||
let bus = event_bus::init_global(DEFAULT_CAPACITY);
|
||||
let _tracing_handle = bus.subscribe(Arc::new(TracingSubscriber));
|
||||
tracing::debug!("[event_bus] global singleton initialized in start_channels");
|
||||
// Note: WebhookRequestSubscriber and ChannelInboundSubscriber are registered
|
||||
// in bootstrap_skill_runtime() (src/core/jsonrpc.rs) to avoid double-registration
|
||||
// when both startup paths run in the same process.
|
||||
|
||||
let provider_runtime_options = providers::ProviderRuntimeOptions {
|
||||
auth_profile_override: None,
|
||||
|
||||
@@ -5,6 +5,7 @@ use super::super::context::{
|
||||
};
|
||||
use super::super::traits;
|
||||
use super::super::Channel;
|
||||
use crate::openhuman::event_bus::{publish_global, DomainEvent};
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
@@ -21,6 +22,9 @@ pub(crate) fn spawn_supervised_listener(
|
||||
|
||||
loop {
|
||||
crate::openhuman::health::mark_component_ok(&component);
|
||||
publish_global(DomainEvent::ChannelConnected {
|
||||
channel: ch.name().to_string(),
|
||||
});
|
||||
let result = ch.listen(tx.clone()).await;
|
||||
|
||||
if tx.is_closed() {
|
||||
@@ -34,12 +38,20 @@ pub(crate) fn spawn_supervised_listener(
|
||||
&component,
|
||||
"listener exited unexpectedly",
|
||||
);
|
||||
publish_global(DomainEvent::ChannelDisconnected {
|
||||
channel: ch.name().to_string(),
|
||||
reason: "exited unexpectedly".to_string(),
|
||||
});
|
||||
// Clean exit — reset backoff since the listener ran successfully
|
||||
backoff = initial_backoff_secs.max(1);
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::error!("Channel {} error: {e}; restarting", ch.name());
|
||||
crate::openhuman::health::mark_component_error(&component, e.to_string());
|
||||
publish_global(DomainEvent::ChannelDisconnected {
|
||||
channel: ch.name().to_string(),
|
||||
reason: e.to_string(),
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -1,7 +1,8 @@
|
||||
//! Domain events for cross-module communication.
|
||||
//!
|
||||
//! All events are lightweight, cloneable snapshots. Heavy data should be
|
||||
//! referenced by ID rather than embedded in the event payload.
|
||||
//! Events carry full payloads so subscribers have everything they need without
|
||||
//! secondary lookups. The broadcast channel clones each event per subscriber,
|
||||
//! which is fine — richness beats round-trips.
|
||||
|
||||
/// Top-level domain event. Non-exhaustive so new variants can be added
|
||||
/// without breaking existing match arms.
|
||||
@@ -35,8 +36,29 @@ pub enum DomainEvent {
|
||||
MemoryRecalled { query: String, hit_count: usize },
|
||||
|
||||
// ── Channels ────────────────────────────────────────────────────────
|
||||
/// An inbound channel message from the transport layer, ready for processing.
|
||||
ChannelInboundMessage {
|
||||
event_name: String,
|
||||
channel: String,
|
||||
message: String,
|
||||
raw_data: serde_json::Value,
|
||||
},
|
||||
/// A message was received on a channel.
|
||||
ChannelMessageReceived { channel: String, sender: String },
|
||||
ChannelMessageReceived {
|
||||
channel: String,
|
||||
sender: String,
|
||||
content: String,
|
||||
thread_ts: Option<String>,
|
||||
},
|
||||
/// A channel message was fully processed (LLM response sent or error).
|
||||
ChannelMessageProcessed {
|
||||
channel: String,
|
||||
sender: String,
|
||||
content: String,
|
||||
response: String,
|
||||
elapsed_ms: u64,
|
||||
success: bool,
|
||||
},
|
||||
/// A channel connected successfully.
|
||||
ChannelConnected { channel: String },
|
||||
/// A channel disconnected.
|
||||
@@ -57,11 +79,17 @@ pub enum DomainEvent {
|
||||
|
||||
// ── Skills ──────────────────────────────────────────────────────────
|
||||
/// A skill was loaded into the runtime.
|
||||
SkillLoaded { skill_id: String },
|
||||
SkillLoaded { skill_id: String, runtime: String },
|
||||
/// A skill was stopped.
|
||||
SkillStopped { skill_id: String },
|
||||
/// A skill failed to start.
|
||||
SkillStartFailed { skill_id: String, error: String },
|
||||
/// A skill tool was executed.
|
||||
SkillExecuted {
|
||||
skill_id: String,
|
||||
tool_name: String,
|
||||
arguments: serde_json::Value,
|
||||
result: Option<String>,
|
||||
success: bool,
|
||||
elapsed_ms: u64,
|
||||
},
|
||||
@@ -81,8 +109,38 @@ pub enum DomainEvent {
|
||||
},
|
||||
|
||||
// ── Webhooks ────────────────────────────────────────────────────────
|
||||
/// An incoming webhook request from the transport layer, ready for routing.
|
||||
WebhookIncomingRequest {
|
||||
request: crate::openhuman::webhooks::WebhookRequest,
|
||||
raw_data: serde_json::Value,
|
||||
},
|
||||
/// A webhook was received and routed to a skill.
|
||||
WebhookReceived { tunnel_id: String, skill_id: String },
|
||||
WebhookReceived {
|
||||
tunnel_id: String,
|
||||
skill_id: String,
|
||||
method: String,
|
||||
path: String,
|
||||
correlation_id: String,
|
||||
},
|
||||
/// A webhook tunnel was registered to a skill.
|
||||
WebhookRegistered {
|
||||
tunnel_id: String,
|
||||
skill_id: String,
|
||||
tunnel_name: Option<String>,
|
||||
},
|
||||
/// A webhook tunnel was unregistered from a skill.
|
||||
WebhookUnregistered { tunnel_id: String, skill_id: String },
|
||||
/// A webhook request was fully processed (includes timing and status).
|
||||
WebhookProcessed {
|
||||
tunnel_id: String,
|
||||
skill_id: String,
|
||||
method: String,
|
||||
path: String,
|
||||
correlation_id: String,
|
||||
status_code: u16,
|
||||
elapsed_ms: u64,
|
||||
error: Option<String>,
|
||||
},
|
||||
|
||||
// ── System lifecycle ────────────────────────────────────────────────
|
||||
/// A system component started up.
|
||||
@@ -103,7 +161,9 @@ impl DomainEvent {
|
||||
|
||||
Self::MemoryStored { .. } | Self::MemoryRecalled { .. } => "memory",
|
||||
|
||||
Self::ChannelMessageReceived { .. }
|
||||
Self::ChannelInboundMessage { .. }
|
||||
| Self::ChannelMessageReceived { .. }
|
||||
| Self::ChannelMessageProcessed { .. }
|
||||
| Self::ChannelConnected { .. }
|
||||
| Self::ChannelDisconnected { .. } => "channel",
|
||||
|
||||
@@ -111,11 +171,18 @@ impl DomainEvent {
|
||||
| Self::CronJobCompleted { .. }
|
||||
| Self::CronDeliveryRequested { .. } => "cron",
|
||||
|
||||
Self::SkillLoaded { .. } | Self::SkillExecuted { .. } => "skill",
|
||||
Self::SkillLoaded { .. }
|
||||
| Self::SkillStopped { .. }
|
||||
| Self::SkillStartFailed { .. }
|
||||
| Self::SkillExecuted { .. } => "skill",
|
||||
|
||||
Self::ToolExecutionStarted { .. } | Self::ToolExecutionCompleted { .. } => "tool",
|
||||
|
||||
Self::WebhookReceived { .. } => "webhook",
|
||||
Self::WebhookIncomingRequest { .. }
|
||||
| Self::WebhookReceived { .. }
|
||||
| Self::WebhookRegistered { .. }
|
||||
| Self::WebhookUnregistered { .. }
|
||||
| Self::WebhookProcessed { .. } => "webhook",
|
||||
|
||||
Self::SystemStartup { .. }
|
||||
| Self::SystemShutdown { .. }
|
||||
@@ -123,3 +190,255 @@ impl DomainEvent {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn all_variants_have_correct_domain() {
|
||||
let cases: Vec<(DomainEvent, &str)> = vec![
|
||||
// Agent
|
||||
(
|
||||
DomainEvent::AgentTurnStarted {
|
||||
session_id: "s".into(),
|
||||
channel: "c".into(),
|
||||
},
|
||||
"agent",
|
||||
),
|
||||
(
|
||||
DomainEvent::AgentTurnCompleted {
|
||||
session_id: "s".into(),
|
||||
text_chars: 0,
|
||||
iterations: 0,
|
||||
},
|
||||
"agent",
|
||||
),
|
||||
(
|
||||
DomainEvent::AgentError {
|
||||
session_id: "s".into(),
|
||||
message: "e".into(),
|
||||
recoverable: false,
|
||||
},
|
||||
"agent",
|
||||
),
|
||||
// Memory
|
||||
(
|
||||
DomainEvent::MemoryStored {
|
||||
key: "k".into(),
|
||||
category: "c".into(),
|
||||
namespace: "n".into(),
|
||||
},
|
||||
"memory",
|
||||
),
|
||||
(
|
||||
DomainEvent::MemoryRecalled {
|
||||
query: "q".into(),
|
||||
hit_count: 0,
|
||||
},
|
||||
"memory",
|
||||
),
|
||||
// Channel
|
||||
(
|
||||
DomainEvent::ChannelInboundMessage {
|
||||
event_name: "telegram:message".into(),
|
||||
channel: "telegram".into(),
|
||||
message: "hi".into(),
|
||||
raw_data: serde_json::Value::Null,
|
||||
},
|
||||
"channel",
|
||||
),
|
||||
(
|
||||
DomainEvent::ChannelMessageReceived {
|
||||
channel: "c".into(),
|
||||
sender: "s".into(),
|
||||
content: "hi".into(),
|
||||
thread_ts: None,
|
||||
},
|
||||
"channel",
|
||||
),
|
||||
(
|
||||
DomainEvent::ChannelMessageProcessed {
|
||||
channel: "c".into(),
|
||||
sender: "s".into(),
|
||||
content: "hi".into(),
|
||||
response: "hello".into(),
|
||||
elapsed_ms: 0,
|
||||
success: true,
|
||||
},
|
||||
"channel",
|
||||
),
|
||||
(
|
||||
DomainEvent::ChannelConnected {
|
||||
channel: "c".into(),
|
||||
},
|
||||
"channel",
|
||||
),
|
||||
(
|
||||
DomainEvent::ChannelDisconnected {
|
||||
channel: "c".into(),
|
||||
reason: "r".into(),
|
||||
},
|
||||
"channel",
|
||||
),
|
||||
// Cron
|
||||
(
|
||||
DomainEvent::CronJobTriggered {
|
||||
job_id: "j".into(),
|
||||
job_type: "t".into(),
|
||||
},
|
||||
"cron",
|
||||
),
|
||||
(
|
||||
DomainEvent::CronJobCompleted {
|
||||
job_id: "j".into(),
|
||||
success: true,
|
||||
},
|
||||
"cron",
|
||||
),
|
||||
(
|
||||
DomainEvent::CronDeliveryRequested {
|
||||
job_id: "j".into(),
|
||||
channel: "c".into(),
|
||||
target: "t".into(),
|
||||
output: "o".into(),
|
||||
},
|
||||
"cron",
|
||||
),
|
||||
// Skill
|
||||
(
|
||||
DomainEvent::SkillLoaded {
|
||||
skill_id: "s".into(),
|
||||
runtime: "quickjs".into(),
|
||||
},
|
||||
"skill",
|
||||
),
|
||||
(
|
||||
DomainEvent::SkillStopped {
|
||||
skill_id: "s".into(),
|
||||
},
|
||||
"skill",
|
||||
),
|
||||
(
|
||||
DomainEvent::SkillStartFailed {
|
||||
skill_id: "s".into(),
|
||||
error: "e".into(),
|
||||
},
|
||||
"skill",
|
||||
),
|
||||
(
|
||||
DomainEvent::SkillExecuted {
|
||||
skill_id: "s".into(),
|
||||
tool_name: "t".into(),
|
||||
arguments: serde_json::Value::Null,
|
||||
result: None,
|
||||
success: true,
|
||||
elapsed_ms: 0,
|
||||
},
|
||||
"skill",
|
||||
),
|
||||
// Tool
|
||||
(
|
||||
DomainEvent::ToolExecutionStarted {
|
||||
tool_name: "t".into(),
|
||||
session_id: "s".into(),
|
||||
},
|
||||
"tool",
|
||||
),
|
||||
(
|
||||
DomainEvent::ToolExecutionCompleted {
|
||||
tool_name: "t".into(),
|
||||
session_id: "s".into(),
|
||||
success: true,
|
||||
elapsed_ms: 0,
|
||||
},
|
||||
"tool",
|
||||
),
|
||||
// Webhook
|
||||
(
|
||||
DomainEvent::WebhookIncomingRequest {
|
||||
request: crate::openhuman::webhooks::WebhookRequest {
|
||||
correlation_id: "c".into(),
|
||||
tunnel_id: "t".into(),
|
||||
tunnel_uuid: "u".into(),
|
||||
tunnel_name: "n".into(),
|
||||
method: "GET".into(),
|
||||
path: "/".into(),
|
||||
headers: Default::default(),
|
||||
query: Default::default(),
|
||||
body: String::new(),
|
||||
},
|
||||
raw_data: serde_json::Value::Null,
|
||||
},
|
||||
"webhook",
|
||||
),
|
||||
(
|
||||
DomainEvent::WebhookReceived {
|
||||
tunnel_id: "t".into(),
|
||||
skill_id: "s".into(),
|
||||
method: "GET".into(),
|
||||
path: "/".into(),
|
||||
correlation_id: "c".into(),
|
||||
},
|
||||
"webhook",
|
||||
),
|
||||
(
|
||||
DomainEvent::WebhookRegistered {
|
||||
tunnel_id: "t".into(),
|
||||
skill_id: "s".into(),
|
||||
tunnel_name: None,
|
||||
},
|
||||
"webhook",
|
||||
),
|
||||
(
|
||||
DomainEvent::WebhookUnregistered {
|
||||
tunnel_id: "t".into(),
|
||||
skill_id: "s".into(),
|
||||
},
|
||||
"webhook",
|
||||
),
|
||||
(
|
||||
DomainEvent::WebhookProcessed {
|
||||
tunnel_id: "t".into(),
|
||||
skill_id: "s".into(),
|
||||
method: "GET".into(),
|
||||
path: "/".into(),
|
||||
correlation_id: "c".into(),
|
||||
status_code: 200,
|
||||
elapsed_ms: 0,
|
||||
error: None,
|
||||
},
|
||||
"webhook",
|
||||
),
|
||||
// System
|
||||
(
|
||||
DomainEvent::SystemStartup {
|
||||
component: "c".into(),
|
||||
},
|
||||
"system",
|
||||
),
|
||||
(
|
||||
DomainEvent::SystemShutdown {
|
||||
component: "c".into(),
|
||||
},
|
||||
"system",
|
||||
),
|
||||
(
|
||||
DomainEvent::HealthChanged {
|
||||
component: "c".into(),
|
||||
healthy: true,
|
||||
},
|
||||
"system",
|
||||
),
|
||||
];
|
||||
|
||||
for (event, expected_domain) in cases {
|
||||
assert_eq!(
|
||||
event.domain(),
|
||||
expected_domain,
|
||||
"Wrong domain for {:?}",
|
||||
std::mem::discriminant(&event)
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,8 @@
|
||||
//! Event bus handlers for the skills domain.
|
||||
//!
|
||||
//! Placeholder for future cross-domain subscribers that react to skill lifecycle
|
||||
//! events (e.g. cascading restarts, dependency tracking, metrics collection).
|
||||
//!
|
||||
//! Skill events are currently consumed by the [`TracingSubscriber`] for
|
||||
//! observability. Add domain-specific handlers here as needed following the
|
||||
//! pattern in `crate::openhuman::cron::bus`.
|
||||
@@ -7,6 +7,7 @@
|
||||
//! - Various operations for skill lifecycle management (`ops.rs`, `registry_ops.rs`)
|
||||
|
||||
pub mod bridge;
|
||||
pub mod bus;
|
||||
pub mod cron_scheduler;
|
||||
pub mod manifest;
|
||||
pub mod ops;
|
||||
|
||||
@@ -30,6 +30,7 @@ pub fn require_engine() -> Result<Arc<RuntimeEngine>, String> {
|
||||
global_engine().ok_or_else(|| "skill runtime not initialized".to_string())
|
||||
}
|
||||
|
||||
use crate::openhuman::event_bus::{publish_global, DomainEvent};
|
||||
use crate::openhuman::memory::{MemoryClient, MemoryClientRef};
|
||||
use crate::openhuman::skills::cron_scheduler::CronScheduler;
|
||||
use crate::openhuman::skills::manifest::SkillManifest;
|
||||
@@ -399,6 +400,7 @@ impl RuntimeEngine {
|
||||
));
|
||||
}
|
||||
|
||||
let runtime_name = manifest.runtime.clone();
|
||||
let config = manifest.to_config();
|
||||
let data_dir = self.skills_data_dir.join(skill_id);
|
||||
|
||||
@@ -452,7 +454,10 @@ impl RuntimeEngine {
|
||||
match current_status {
|
||||
SkillStatus::Running => {
|
||||
self.emit_status_change(&skill_id_owned);
|
||||
|
||||
publish_global(DomainEvent::SkillLoaded {
|
||||
skill_id: skill_id_owned.clone(),
|
||||
runtime: runtime_name.clone(),
|
||||
});
|
||||
return Ok(instance.snapshot());
|
||||
}
|
||||
SkillStatus::Error => {
|
||||
@@ -464,6 +469,10 @@ impl RuntimeEngine {
|
||||
// Don't unregister — keep the skill with Error status so the
|
||||
// UI can see what happened and allow restart.
|
||||
self.emit_status_change(&skill_id_owned);
|
||||
publish_global(DomainEvent::SkillStartFailed {
|
||||
skill_id: skill_id_owned.clone(),
|
||||
error: error_msg.clone(),
|
||||
});
|
||||
return Err(format!(
|
||||
"Skill '{}' failed to start: {}",
|
||||
skill_id_owned, error_msg
|
||||
@@ -507,6 +516,9 @@ impl RuntimeEngine {
|
||||
self.cron_scheduler.unregister_all_for_skill(skill_id);
|
||||
self.webhook_router.unregister_skill(skill_id);
|
||||
self.emit_status_change(skill_id);
|
||||
publish_global(DomainEvent::SkillStopped {
|
||||
skill_id: skill_id.to_string(),
|
||||
});
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -549,9 +561,24 @@ impl RuntimeEngine {
|
||||
tool_name: &str,
|
||||
arguments: serde_json::Value,
|
||||
) -> Result<ToolResult, String> {
|
||||
self.registry
|
||||
.call_tool(skill_id, tool_name, arguments)
|
||||
.await
|
||||
let started_at = std::time::Instant::now();
|
||||
let result = self
|
||||
.registry
|
||||
.call_tool(skill_id, tool_name, arguments.clone())
|
||||
.await;
|
||||
let result_text = match &result {
|
||||
Ok(r) => Some(r.output()),
|
||||
Err(e) => Some(e.clone()),
|
||||
};
|
||||
publish_global(DomainEvent::SkillExecuted {
|
||||
skill_id: skill_id.to_string(),
|
||||
tool_name: tool_name.to_string(),
|
||||
arguments,
|
||||
result: result_text,
|
||||
success: result.is_ok(),
|
||||
elapsed_ms: started_at.elapsed().as_millis() as u64,
|
||||
});
|
||||
result
|
||||
}
|
||||
|
||||
/// Call a tool from inside a running skill. Enforces self-only invocation.
|
||||
@@ -565,16 +592,31 @@ impl RuntimeEngine {
|
||||
tool_name: &str,
|
||||
arguments: serde_json::Value,
|
||||
) -> Result<ToolResult, String> {
|
||||
self.registry
|
||||
let started_at = std::time::Instant::now();
|
||||
let result = self
|
||||
.registry
|
||||
.call_tool_scoped(
|
||||
ToolCallOrigin::SkillSelf {
|
||||
skill_id: caller_skill_id.to_string(),
|
||||
},
|
||||
target_skill_id,
|
||||
tool_name,
|
||||
arguments,
|
||||
arguments.clone(),
|
||||
)
|
||||
.await
|
||||
.await;
|
||||
let result_text = match &result {
|
||||
Ok(r) => Some(r.output()),
|
||||
Err(e) => Some(e.clone()),
|
||||
};
|
||||
publish_global(DomainEvent::SkillExecuted {
|
||||
skill_id: target_skill_id.to_string(),
|
||||
tool_name: tool_name.to_string(),
|
||||
arguments,
|
||||
result: result_text,
|
||||
success: result.is_ok(),
|
||||
elapsed_ms: started_at.elapsed().as_millis() as u64,
|
||||
});
|
||||
result
|
||||
}
|
||||
|
||||
/// Broadcast a server event to all running skills.
|
||||
|
||||
@@ -1,7 +1,8 @@
|
||||
//! Socket.IO event routing and protocol handlers.
|
||||
//!
|
||||
//! Dispatches incoming Socket.IO events to the appropriate handler:
|
||||
//! webhook tunnel requests, channel inbound messages, or generic event logging.
|
||||
//! Thin transport layer: parses incoming Socket.IO events and publishes them
|
||||
//! to the event bus for domain-specific handling. Webhook routing lives in
|
||||
//! `webhooks::bus`, channel inbound processing lives in `channels::bus`.
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
@@ -9,6 +10,7 @@ use serde_json::json;
|
||||
use tokio::sync::mpsc;
|
||||
|
||||
use crate::api::models::socket::ConnectionStatus;
|
||||
use crate::openhuman::event_bus::{publish_global, DomainEvent};
|
||||
use crate::openhuman::webhooks::WebhookRequest;
|
||||
|
||||
use super::manager::{emit_server_event, emit_state_change, SharedState};
|
||||
@@ -21,7 +23,7 @@ use super::manager::{emit_server_event, emit_state_change, SharedState};
|
||||
pub(super) fn handle_sio_event(
|
||||
event_name: &str,
|
||||
data: serde_json::Value,
|
||||
emit_tx: &mpsc::UnboundedSender<String>,
|
||||
_emit_tx: &mpsc::UnboundedSender<String>,
|
||||
shared: &Arc<SharedState>,
|
||||
) {
|
||||
// Log every incoming event for observability.
|
||||
@@ -47,26 +49,102 @@ pub(super) fn handle_sio_event(
|
||||
*shared.status.write() = ConnectionStatus::Error;
|
||||
emit_state_change(shared);
|
||||
}
|
||||
// Webhook tunnel — route to owning skill and relay response
|
||||
// Webhook tunnel — publish to event bus for routing by WebhookRequestSubscriber
|
||||
"webhook:request" => {
|
||||
log::info!("[socket] Routing webhook:request to handler");
|
||||
let shared = Arc::clone(shared);
|
||||
let tx = emit_tx.clone();
|
||||
tokio::spawn(async move {
|
||||
handle_webhook_request(&shared, data, &tx).await;
|
||||
});
|
||||
log::info!("[socket] Publishing webhook:request to event bus");
|
||||
match serde_json::from_value::<WebhookRequest>(data.clone()) {
|
||||
Ok(request) => {
|
||||
publish_global(DomainEvent::WebhookIncomingRequest {
|
||||
request,
|
||||
raw_data: data,
|
||||
});
|
||||
}
|
||||
Err(e) => {
|
||||
log::error!("[socket] Failed to parse webhook:request payload: {e}");
|
||||
// Publish with a minimal request so the subscriber can still
|
||||
// emit an error response. Build a request from what we can parse.
|
||||
let cid = data
|
||||
.get("correlationId")
|
||||
.and_then(|v| v.as_str())
|
||||
.unwrap_or("unknown")
|
||||
.to_string();
|
||||
let _tunnel_uuid = data
|
||||
.get("tunnelUuid")
|
||||
.and_then(|v| v.as_str())
|
||||
.unwrap_or("")
|
||||
.to_string();
|
||||
|
||||
// Record parse error in router debug log if available
|
||||
if let Some(router) = shared.webhook_router.read().clone() {
|
||||
router.record_parse_error(
|
||||
cid.clone(),
|
||||
data.get("tunnelUuid")
|
||||
.and_then(|v| v.as_str())
|
||||
.map(|v| v.to_string()),
|
||||
data.get("method")
|
||||
.and_then(|v| v.as_str())
|
||||
.map(|v| v.to_string()),
|
||||
data.get("path")
|
||||
.and_then(|v| v.as_str())
|
||||
.map(|v| v.to_string()),
|
||||
data.clone(),
|
||||
format!("bad request: {e}"),
|
||||
);
|
||||
}
|
||||
|
||||
// Emit error response directly via socket manager
|
||||
if let Some(mgr) = crate::openhuman::socket::global_socket_manager() {
|
||||
let err_json = json!({ "error": format!("Bad request: {e}") });
|
||||
let body = base64_encode(&err_json.to_string());
|
||||
let response_data = json!({
|
||||
"correlationId": cid,
|
||||
"statusCode": 400,
|
||||
"headers": {},
|
||||
"body": body,
|
||||
});
|
||||
let mgr = mgr.clone();
|
||||
tokio::spawn(async move {
|
||||
if let Err(e) = mgr.emit("webhook:response", response_data).await {
|
||||
log::error!("[socket] Failed to emit webhook error response: {e}");
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
// Any event ending with ":message" is treated as an inbound channel
|
||||
// message that triggers the agent loop. This covers channel:message,
|
||||
// telegram:message, discord:message, slack:message, and any future
|
||||
// channel integration without requiring code changes.
|
||||
// Channel inbound message — publish to event bus for ChannelInboundSubscriber
|
||||
_ if event_name.ends_with(":message") => {
|
||||
log::info!(
|
||||
"[socket] Inbound channel message via '{}' — triggering agent loop",
|
||||
"[socket] Publishing inbound channel message '{}' to event bus",
|
||||
event_name
|
||||
);
|
||||
tokio::spawn(async move {
|
||||
handle_channel_inbound_message(data).await;
|
||||
|
||||
let channel = data
|
||||
.get("channel")
|
||||
.and_then(|v| v.as_str())
|
||||
.unwrap_or("")
|
||||
.to_string();
|
||||
let message = data
|
||||
.get("message")
|
||||
.and_then(|v| v.as_str())
|
||||
.unwrap_or("")
|
||||
.trim()
|
||||
.to_string();
|
||||
|
||||
if channel.is_empty() {
|
||||
log::warn!("[socket] channel:message missing 'channel' field");
|
||||
return;
|
||||
}
|
||||
if message.is_empty() {
|
||||
log::debug!("[socket] channel:message empty or missing 'message'");
|
||||
return;
|
||||
}
|
||||
|
||||
publish_global(DomainEvent::ChannelInboundMessage {
|
||||
event_name: event_name.to_string(),
|
||||
channel,
|
||||
message,
|
||||
raw_data: data,
|
||||
});
|
||||
}
|
||||
_ => {
|
||||
@@ -76,387 +154,11 @@ pub(super) fn handle_sio_event(
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Webhook tunnel handler
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/// Handle an incoming `webhook:request` event from the backend.
|
||||
///
|
||||
/// Routes the request to the owning skill via the WebhookRouter, waits for the
|
||||
/// skill's response, and emits `webhook:response` back through the socket.
|
||||
async fn handle_webhook_request(
|
||||
shared: &SharedState,
|
||||
data: serde_json::Value,
|
||||
emit_tx: &mpsc::UnboundedSender<String>,
|
||||
) {
|
||||
let request: WebhookRequest = match serde_json::from_value(data.clone()) {
|
||||
Ok(r) => r,
|
||||
Err(e) => {
|
||||
log::error!("[socket] Failed to parse webhook:request payload: {e}");
|
||||
let cid = data
|
||||
.get("correlationId")
|
||||
.and_then(|v| v.as_str())
|
||||
.unwrap_or("unknown")
|
||||
.to_string();
|
||||
if let Some(router) = shared.webhook_router.read().clone() {
|
||||
router.record_parse_error(
|
||||
cid.clone(),
|
||||
data.get("tunnelUuid")
|
||||
.and_then(|value| value.as_str())
|
||||
.map(|value| value.to_string()),
|
||||
data.get("method")
|
||||
.and_then(|value| value.as_str())
|
||||
.map(|value| value.to_string()),
|
||||
data.get("path")
|
||||
.and_then(|value| value.as_str())
|
||||
.map(|value| value.to_string()),
|
||||
data.clone(),
|
||||
format!("bad request: {e}"),
|
||||
);
|
||||
}
|
||||
emit_via_channel(
|
||||
emit_tx,
|
||||
"webhook:response",
|
||||
json!({
|
||||
"correlationId": cid,
|
||||
"statusCode": 400,
|
||||
"headers": {},
|
||||
"body": base64_encode(&format!(
|
||||
"{{\"error\":\"Bad request: {}\"}}",
|
||||
e.to_string().replace('"', "\\\"")
|
||||
)),
|
||||
}),
|
||||
);
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
let correlation_id = request.correlation_id.clone();
|
||||
let tunnel_uuid = request.tunnel_uuid.clone();
|
||||
let tunnel_name = request.tunnel_name.clone();
|
||||
let method = request.method.clone();
|
||||
let path = request.path.clone();
|
||||
|
||||
log::info!(
|
||||
"[socket] webhook:request {} {} (tunnel={}, correlationId={})",
|
||||
method,
|
||||
path,
|
||||
tunnel_uuid,
|
||||
correlation_id,
|
||||
);
|
||||
|
||||
let router = shared.webhook_router.read().clone();
|
||||
let registration = router.as_ref().and_then(|r| r.registration(&tunnel_uuid));
|
||||
let skill_id = registration.as_ref().and_then(|registration| {
|
||||
(registration.target_kind == "skill").then(|| registration.skill_id.clone())
|
||||
});
|
||||
if let Some(router) = router.as_ref() {
|
||||
router.record_request(&request, skill_id.clone());
|
||||
}
|
||||
|
||||
let (response, resolved_skill_id, response_error) = match registration.as_ref() {
|
||||
Some(registration) if registration.target_kind == "echo" => (
|
||||
crate::openhuman::webhooks::ops::build_echo_response(&request),
|
||||
Some("echo".to_string()),
|
||||
None,
|
||||
),
|
||||
Some(registration) if registration.target_kind == "channel" => (
|
||||
crate::openhuman::webhooks::WebhookResponseData {
|
||||
correlation_id: correlation_id.clone(),
|
||||
status_code: 501,
|
||||
headers: std::collections::HashMap::new(),
|
||||
body: base64_encode(&format!(
|
||||
"{{\"error\":\"channel webhook target '{}' is not implemented in this runtime yet\"}}",
|
||||
registration.skill_id.replace('"', "\\\"")
|
||||
)),
|
||||
},
|
||||
Some(registration.skill_id.clone()),
|
||||
Some("channel webhook target not implemented".to_string()),
|
||||
),
|
||||
Some(registration) if registration.target_kind == "skill" => {
|
||||
let sid = registration.skill_id.clone();
|
||||
log::debug!("[socket] webhook:request routed to skill '{}'", sid);
|
||||
|
||||
let registry = crate::openhuman::skills::global_engine()
|
||||
.map(|e| e.registry());
|
||||
match registry {
|
||||
Some(registry) => {
|
||||
let result = registry
|
||||
.send_webhook_request(
|
||||
&sid,
|
||||
correlation_id.clone(),
|
||||
request.method.clone(),
|
||||
request.path.clone(),
|
||||
request.headers.clone(),
|
||||
request.query.clone(),
|
||||
request.body.clone(),
|
||||
request.tunnel_id.clone(),
|
||||
request.tunnel_name.clone(),
|
||||
)
|
||||
.await;
|
||||
|
||||
match result {
|
||||
Ok(resp) => (resp, Some(sid), None),
|
||||
Err(e) => {
|
||||
log::warn!("[socket] Skill webhook handler error: {}", e);
|
||||
(
|
||||
crate::openhuman::webhooks::WebhookResponseData {
|
||||
correlation_id: correlation_id.clone(),
|
||||
status_code: 500,
|
||||
headers: std::collections::HashMap::new(),
|
||||
body: base64_encode(&format!(
|
||||
"{{\"error\":\"Skill handler error: {}\"}}",
|
||||
e.replace('"', "\\\"")
|
||||
)),
|
||||
},
|
||||
Some(sid),
|
||||
Some(e),
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
None => {
|
||||
log::warn!("[socket] No skill registry available for webhook");
|
||||
(
|
||||
crate::openhuman::webhooks::WebhookResponseData {
|
||||
correlation_id: correlation_id.clone(),
|
||||
status_code: 503,
|
||||
headers: std::collections::HashMap::new(),
|
||||
body: base64_encode("{\"error\":\"Runtime not ready\"}"),
|
||||
},
|
||||
None,
|
||||
Some("runtime not ready".to_string()),
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
Some(registration) => (
|
||||
crate::openhuman::webhooks::WebhookResponseData {
|
||||
correlation_id: correlation_id.clone(),
|
||||
status_code: 400,
|
||||
headers: std::collections::HashMap::new(),
|
||||
body: base64_encode(&format!(
|
||||
"{{\"error\":\"unknown webhook target kind '{}'\"}}",
|
||||
registration.target_kind.replace('"', "\\\"")
|
||||
)),
|
||||
},
|
||||
Some(registration.skill_id.clone()),
|
||||
Some("unknown webhook target kind".to_string()),
|
||||
),
|
||||
None => {
|
||||
log::debug!(
|
||||
"[socket] No skill registered for tunnel {}",
|
||||
tunnel_uuid,
|
||||
);
|
||||
(
|
||||
crate::openhuman::webhooks::WebhookResponseData {
|
||||
correlation_id: correlation_id.clone(),
|
||||
status_code: 404,
|
||||
headers: std::collections::HashMap::new(),
|
||||
body: base64_encode("{\"error\":\"No handler registered for this tunnel\"}"),
|
||||
},
|
||||
None,
|
||||
Some("no handler registered for this tunnel".to_string()),
|
||||
)
|
||||
}
|
||||
};
|
||||
|
||||
if let Some(router) = router.as_ref() {
|
||||
router.record_response(
|
||||
&request,
|
||||
&response,
|
||||
resolved_skill_id.clone(),
|
||||
response_error.clone(),
|
||||
);
|
||||
}
|
||||
|
||||
emit_via_channel(
|
||||
emit_tx,
|
||||
"webhook:response",
|
||||
json!({
|
||||
"correlationId": response.correlation_id,
|
||||
"statusCode": response.status_code,
|
||||
"headers": response.headers,
|
||||
"body": response.body,
|
||||
}),
|
||||
);
|
||||
|
||||
log::info!(
|
||||
"[socket] webhook activity: {} {} → status={}, skill={:?}, tunnel={}",
|
||||
method,
|
||||
path,
|
||||
response.status_code,
|
||||
resolved_skill_id,
|
||||
tunnel_name,
|
||||
);
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Channel inbound message → agent loop → reply
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/// Handle an inbound message from a channel (Telegram, Discord, etc.).
|
||||
///
|
||||
/// Runs the agent inference loop via `web::start_chat` and sends the response
|
||||
/// back to the originating channel via the REST API.
|
||||
async fn handle_channel_inbound_message(data: serde_json::Value) {
|
||||
let channel = match data.get("channel").and_then(|v| v.as_str()) {
|
||||
Some(c) => c.to_string(),
|
||||
None => {
|
||||
log::warn!("[channel-inbound] channel:message missing 'channel' field");
|
||||
return;
|
||||
}
|
||||
};
|
||||
let message = match data.get("message").and_then(|v| v.as_str()) {
|
||||
Some(m) if !m.trim().is_empty() => m.trim().to_string(),
|
||||
_ => {
|
||||
log::debug!("[channel-inbound] channel:message empty or missing 'message'");
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
log::info!(
|
||||
"[channel-inbound] received message from channel='{}' len={}",
|
||||
channel,
|
||||
message.len()
|
||||
);
|
||||
|
||||
let thread_id = format!("channel:{}", channel);
|
||||
let client_id = "inbound".to_string();
|
||||
|
||||
let mut event_rx = crate::openhuman::channels::providers::web::subscribe_web_channel_events();
|
||||
|
||||
let request_id = match crate::openhuman::channels::providers::web::start_chat(
|
||||
&client_id, &thread_id, &message, None, None,
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(rid) => {
|
||||
log::debug!(
|
||||
"[channel-inbound] agent started request_id={} thread={}",
|
||||
rid,
|
||||
thread_id
|
||||
);
|
||||
rid
|
||||
}
|
||||
Err(err) => {
|
||||
log::error!("[channel-inbound] start_chat failed: {}", err);
|
||||
send_channel_reply(
|
||||
&channel,
|
||||
&format!("Sorry, I couldn't process your message: {err}"),
|
||||
)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
let timeout = tokio::time::Duration::from_secs(180);
|
||||
let deadline = tokio::time::Instant::now() + timeout;
|
||||
|
||||
loop {
|
||||
tokio::select! {
|
||||
event = event_rx.recv() => {
|
||||
match event {
|
||||
Ok(ev) if ev.request_id == request_id => {
|
||||
if ev.event == "chat_done" || ev.event == "chat:done" {
|
||||
let reply = ev.full_response.unwrap_or_default();
|
||||
if reply.trim().is_empty() {
|
||||
log::warn!("[channel-inbound] agent returned empty response");
|
||||
return;
|
||||
}
|
||||
log::info!(
|
||||
"[channel-inbound] agent done, replying to channel='{}' len={}",
|
||||
channel,
|
||||
reply.len()
|
||||
);
|
||||
send_channel_reply(&channel, &reply).await;
|
||||
return;
|
||||
}
|
||||
if ev.event == "chat_error" || ev.event == "chat:error" {
|
||||
let err_msg = ev.message.unwrap_or_else(|| "unknown error".to_string());
|
||||
log::error!("[channel-inbound] agent error: {}", err_msg);
|
||||
send_channel_reply(
|
||||
&channel,
|
||||
&format!("Sorry, I encountered an error: {err_msg}"),
|
||||
)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
}
|
||||
Ok(_) => {}
|
||||
Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
|
||||
log::warn!("[channel-inbound] event bus lagged, skipped {} events", n);
|
||||
}
|
||||
Err(tokio::sync::broadcast::error::RecvError::Closed) => {
|
||||
log::error!("[channel-inbound] event bus closed unexpectedly");
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
_ = tokio::time::sleep_until(deadline) => {
|
||||
log::error!("[channel-inbound] agent timed out after {}s", timeout.as_secs());
|
||||
send_channel_reply(&channel, "Sorry, the request timed out.").await;
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Send a text reply back to a channel via the backend REST API.
|
||||
async fn send_channel_reply(channel: &str, text: &str) {
|
||||
let config = match crate::openhuman::config::rpc::load_config_with_timeout().await {
|
||||
Ok(c) => c,
|
||||
Err(e) => {
|
||||
log::error!("[channel-inbound] failed to load config: {}", e);
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
let api_url = crate::api::config::effective_api_url(&config.api_url);
|
||||
let jwt = match crate::api::jwt::get_session_token(&config) {
|
||||
Ok(Some(t)) => t,
|
||||
Ok(None) => {
|
||||
log::error!("[channel-inbound] no session JWT — cannot reply");
|
||||
return;
|
||||
}
|
||||
Err(e) => {
|
||||
log::error!("[channel-inbound] failed to get session token: {}", e);
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
let client = match crate::api::rest::BackendOAuthClient::new(&api_url) {
|
||||
Ok(c) => c,
|
||||
Err(e) => {
|
||||
log::error!("[channel-inbound] failed to create API client: {}", e);
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
let body = json!({ "text": text });
|
||||
match client.send_channel_message(channel, &jwt, body).await {
|
||||
Ok(resp) => {
|
||||
log::info!(
|
||||
"[channel-inbound] reply sent to channel='{}' response={:?}",
|
||||
channel,
|
||||
resp
|
||||
);
|
||||
}
|
||||
Err(e) => {
|
||||
log::error!(
|
||||
"[channel-inbound] failed to send reply to channel='{}': {}",
|
||||
channel,
|
||||
e
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Utility functions
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/// Base64-encode a string (for webhook response bodies).
|
||||
/// Base64-encode a string (for webhook error response bodies).
|
||||
fn base64_encode(input: &str) -> String {
|
||||
use base64::Engine;
|
||||
base64::engine::general_purpose::STANDARD.encode(input.as_bytes())
|
||||
|
||||
@@ -0,0 +1,237 @@
|
||||
//! Event bus handlers for the webhook domain.
|
||||
//!
|
||||
//! The [`WebhookRequestSubscriber`] handles incoming webhook requests published
|
||||
//! by the socket transport layer. It routes each request to the owning skill (or
|
||||
//! echo target), waits for the response, and emits it back through the socket.
|
||||
//! This decouples the socket module from webhook routing logic.
|
||||
|
||||
use crate::openhuman::event_bus::{publish_global, DomainEvent, EventHandler};
|
||||
use crate::openhuman::socket::global_socket_manager;
|
||||
use async_trait::async_trait;
|
||||
use serde_json::json;
|
||||
use std::time::Instant;
|
||||
|
||||
/// Base64-encode a string (for webhook response bodies).
|
||||
fn base64_encode(input: &str) -> String {
|
||||
use base64::Engine;
|
||||
base64::engine::general_purpose::STANDARD.encode(input.as_bytes())
|
||||
}
|
||||
|
||||
/// Build a base64-encoded JSON error body using proper serialization.
|
||||
fn error_body(message: &str) -> String {
|
||||
let obj = serde_json::json!({ "error": message });
|
||||
base64_encode(&obj.to_string())
|
||||
}
|
||||
|
||||
/// Subscribes to `WebhookIncomingRequest` events and handles the full routing
|
||||
/// flow: lookup tunnel → dispatch to skill/echo → emit response via socket.
|
||||
pub struct WebhookRequestSubscriber;
|
||||
|
||||
impl WebhookRequestSubscriber {
|
||||
pub fn new() -> Self {
|
||||
Self
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl EventHandler for WebhookRequestSubscriber {
|
||||
fn name(&self) -> &str {
|
||||
"webhook::request_handler"
|
||||
}
|
||||
|
||||
fn domains(&self) -> Option<&[&str]> {
|
||||
Some(&["webhook"])
|
||||
}
|
||||
|
||||
async fn handle(&self, event: &DomainEvent) {
|
||||
let DomainEvent::WebhookIncomingRequest {
|
||||
request,
|
||||
raw_data: _,
|
||||
} = event
|
||||
else {
|
||||
return;
|
||||
};
|
||||
|
||||
let started_at = Instant::now();
|
||||
|
||||
let correlation_id = request.correlation_id.clone();
|
||||
let tunnel_uuid = request.tunnel_uuid.clone();
|
||||
let tunnel_name = request.tunnel_name.clone();
|
||||
let method = request.method.clone();
|
||||
let path = request.path.clone();
|
||||
|
||||
tracing::info!(
|
||||
"[webhook] incoming request {} {} (tunnel={}, correlationId={})",
|
||||
method,
|
||||
path,
|
||||
tunnel_uuid,
|
||||
correlation_id,
|
||||
);
|
||||
|
||||
// Get the webhook router from the skill engine
|
||||
let router = crate::openhuman::skills::global_engine().map(|e| e.webhook_router());
|
||||
|
||||
let registration = router.as_ref().and_then(|r| r.registration(&tunnel_uuid));
|
||||
let skill_id = registration
|
||||
.as_ref()
|
||||
.and_then(|reg| (reg.target_kind == "skill").then(|| reg.skill_id.clone()));
|
||||
if let Some(ref router) = router {
|
||||
router.record_request(request, skill_id.clone());
|
||||
}
|
||||
|
||||
let (response, resolved_skill_id, response_error) = match registration.as_ref() {
|
||||
Some(reg) if reg.target_kind == "echo" => (
|
||||
crate::openhuman::webhooks::ops::build_echo_response(request),
|
||||
Some("echo".to_string()),
|
||||
None,
|
||||
),
|
||||
Some(reg) if reg.target_kind == "channel" => (
|
||||
crate::openhuman::webhooks::WebhookResponseData {
|
||||
correlation_id: correlation_id.clone(),
|
||||
status_code: 501,
|
||||
headers: std::collections::HashMap::new(),
|
||||
body: error_body(&format!(
|
||||
"channel webhook target '{}' is not implemented in this runtime yet",
|
||||
reg.skill_id
|
||||
)),
|
||||
},
|
||||
Some(reg.skill_id.clone()),
|
||||
Some("channel webhook target not implemented".to_string()),
|
||||
),
|
||||
Some(reg) if reg.target_kind == "skill" => {
|
||||
let sid = reg.skill_id.clone();
|
||||
tracing::debug!("[webhook] request routed to skill '{}'", sid);
|
||||
|
||||
let registry = crate::openhuman::skills::global_engine().map(|e| e.registry());
|
||||
match registry {
|
||||
Some(registry) => {
|
||||
let result = registry
|
||||
.send_webhook_request(
|
||||
&sid,
|
||||
correlation_id.clone(),
|
||||
request.method.clone(),
|
||||
request.path.clone(),
|
||||
request.headers.clone(),
|
||||
request.query.clone(),
|
||||
request.body.clone(),
|
||||
request.tunnel_id.clone(),
|
||||
request.tunnel_name.clone(),
|
||||
)
|
||||
.await;
|
||||
|
||||
match result {
|
||||
Ok(resp) => (resp, Some(sid), None),
|
||||
Err(e) => {
|
||||
tracing::warn!("[webhook] skill handler error: {}", e);
|
||||
(
|
||||
crate::openhuman::webhooks::WebhookResponseData {
|
||||
correlation_id: correlation_id.clone(),
|
||||
status_code: 500,
|
||||
headers: std::collections::HashMap::new(),
|
||||
body: error_body(&format!("Skill handler error: {}", e)),
|
||||
},
|
||||
Some(sid),
|
||||
Some(e),
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
None => {
|
||||
tracing::warn!("[webhook] no skill registry available");
|
||||
(
|
||||
crate::openhuman::webhooks::WebhookResponseData {
|
||||
correlation_id: correlation_id.clone(),
|
||||
status_code: 503,
|
||||
headers: std::collections::HashMap::new(),
|
||||
body: error_body("Runtime not ready"),
|
||||
},
|
||||
None,
|
||||
Some("runtime not ready".to_string()),
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
Some(reg) => (
|
||||
crate::openhuman::webhooks::WebhookResponseData {
|
||||
correlation_id: correlation_id.clone(),
|
||||
status_code: 400,
|
||||
headers: std::collections::HashMap::new(),
|
||||
body: error_body(&format!(
|
||||
"unknown webhook target kind '{}'",
|
||||
reg.target_kind
|
||||
)),
|
||||
},
|
||||
Some(reg.skill_id.clone()),
|
||||
Some("unknown webhook target kind".to_string()),
|
||||
),
|
||||
None => {
|
||||
tracing::debug!("[webhook] no handler registered for tunnel {}", tunnel_uuid,);
|
||||
(
|
||||
crate::openhuman::webhooks::WebhookResponseData {
|
||||
correlation_id: correlation_id.clone(),
|
||||
status_code: 404,
|
||||
headers: std::collections::HashMap::new(),
|
||||
body: error_body("No handler registered for this tunnel"),
|
||||
},
|
||||
None,
|
||||
Some("no handler registered for this tunnel".to_string()),
|
||||
)
|
||||
}
|
||||
};
|
||||
|
||||
// Record in debug log
|
||||
if let Some(ref router) = router {
|
||||
router.record_response(
|
||||
request,
|
||||
&response,
|
||||
resolved_skill_id.clone(),
|
||||
response_error.clone(),
|
||||
);
|
||||
}
|
||||
|
||||
// Publish notification events
|
||||
if let Some(ref sid) = resolved_skill_id {
|
||||
publish_global(DomainEvent::WebhookReceived {
|
||||
tunnel_id: tunnel_uuid.clone(),
|
||||
skill_id: sid.clone(),
|
||||
method: method.clone(),
|
||||
path: path.clone(),
|
||||
correlation_id: correlation_id.clone(),
|
||||
});
|
||||
}
|
||||
publish_global(DomainEvent::WebhookProcessed {
|
||||
tunnel_id: tunnel_uuid.clone(),
|
||||
skill_id: resolved_skill_id.clone().unwrap_or_default(),
|
||||
method: method.clone(),
|
||||
path: path.clone(),
|
||||
correlation_id: correlation_id.clone(),
|
||||
status_code: response.status_code,
|
||||
elapsed_ms: started_at.elapsed().as_millis() as u64,
|
||||
error: response_error.clone(),
|
||||
});
|
||||
|
||||
// Emit response back through the socket
|
||||
if let Some(mgr) = global_socket_manager() {
|
||||
let response_data = json!({
|
||||
"correlationId": response.correlation_id,
|
||||
"statusCode": response.status_code,
|
||||
"headers": response.headers,
|
||||
"body": response.body,
|
||||
});
|
||||
if let Err(e) = mgr.emit("webhook:response", response_data).await {
|
||||
tracing::error!("[webhook] failed to emit response via socket: {}", e);
|
||||
}
|
||||
} else {
|
||||
tracing::error!("[webhook] no socket manager available to emit response");
|
||||
}
|
||||
|
||||
tracing::info!(
|
||||
"[webhook] {} {} → status={}, skill={:?}, tunnel={}",
|
||||
method,
|
||||
path,
|
||||
response.status_code,
|
||||
resolved_skill_id,
|
||||
tunnel_name,
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -4,6 +4,7 @@
|
||||
//! appropriate skill. The backend manages tunnel provisioning (ngrok, cloudflare,
|
||||
//! etc.); this module handles the client-side routing and skill dispatch.
|
||||
|
||||
pub mod bus;
|
||||
pub mod ops;
|
||||
pub mod router;
|
||||
mod schemas;
|
||||
|
||||
@@ -4,6 +4,7 @@ use super::types::{
|
||||
TunnelRegistration, WebhookDebugEvent, WebhookDebugLogEntry, WebhookRequest,
|
||||
WebhookResponseData,
|
||||
};
|
||||
use crate::openhuman::event_bus::{publish_global, DomainEvent};
|
||||
use log::{debug, error, warn};
|
||||
use once_cell::sync::Lazy;
|
||||
use serde::{Deserialize, Serialize};
|
||||
@@ -140,6 +141,7 @@ impl WebhookRouter {
|
||||
tunnel_uuid, target_kind, skill_id
|
||||
);
|
||||
|
||||
let tunnel_name_clone = tunnel_name.clone();
|
||||
routes.insert(
|
||||
tunnel_uuid.to_string(),
|
||||
TunnelRegistration {
|
||||
@@ -154,6 +156,13 @@ impl WebhookRouter {
|
||||
drop(routes);
|
||||
self.publish_event("registration_changed", None, Some(tunnel_uuid.to_string()));
|
||||
self.persist();
|
||||
|
||||
publish_global(DomainEvent::WebhookRegistered {
|
||||
tunnel_id: tunnel_uuid.to_string(),
|
||||
skill_id: skill_id.to_string(),
|
||||
tunnel_name: tunnel_name_clone,
|
||||
});
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -183,6 +192,12 @@ impl WebhookRouter {
|
||||
drop(routes);
|
||||
self.publish_event("registration_changed", None, Some(tunnel_uuid.to_string()));
|
||||
self.persist();
|
||||
|
||||
publish_global(DomainEvent::WebhookUnregistered {
|
||||
tunnel_id: tunnel_uuid.to_string(),
|
||||
skill_id: skill_id.to_string(),
|
||||
});
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -196,18 +211,30 @@ impl WebhookRouter {
|
||||
}
|
||||
};
|
||||
|
||||
let before = routes.len();
|
||||
routes.retain(|_, reg| reg.skill_id != skill_id);
|
||||
let removed = before - routes.len();
|
||||
let removed_tunnels: Vec<String> = routes
|
||||
.iter()
|
||||
.filter(|(_, reg)| reg.skill_id == skill_id)
|
||||
.map(|(uuid, _)| uuid.clone())
|
||||
.collect();
|
||||
|
||||
if removed > 0 {
|
||||
routes.retain(|_, reg| reg.skill_id != skill_id);
|
||||
|
||||
if !removed_tunnels.is_empty() {
|
||||
debug!(
|
||||
"[webhooks] Unregistered {} tunnel(s) for skill '{}'",
|
||||
removed, skill_id
|
||||
removed_tunnels.len(),
|
||||
skill_id
|
||||
);
|
||||
drop(routes);
|
||||
self.publish_event("registration_changed", None, None);
|
||||
self.persist();
|
||||
|
||||
for tunnel_id in removed_tunnels {
|
||||
publish_global(DomainEvent::WebhookUnregistered {
|
||||
tunnel_id,
|
||||
skill_id: skill_id.to_string(),
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user