diff --git a/app/src/AppRoutes.tsx b/app/src/AppRoutes.tsx index 6f2b8b666..4f0c33b46 100644 --- a/app/src/AppRoutes.tsx +++ b/app/src/AppRoutes.tsx @@ -10,6 +10,7 @@ import Intelligence from './pages/Intelligence'; import Invites from './pages/Invites'; import Settings from './pages/Settings'; import Skills from './pages/Skills'; +import Webhooks from './pages/Webhooks'; import Welcome from './pages/Welcome'; const AppRoutes = () => { @@ -88,6 +89,15 @@ const AppRoutes = () => { } /> + + + + } + /> + { const [apiUrl, setApiUrl] = useState(''); const [defaultModel, setDefaultModel] = useState(''); const [defaultTemp, setDefaultTemp] = useState('0.7'); - const [tunnelProvider, setTunnelProvider] = useState('none'); - const [cloudflareToken, setCloudflareToken] = useState(''); - const [ngrokToken, setNgrokToken] = useState(''); - const [tailscaleHostname, setTailscaleHostname] = useState(''); - const [customCommand, setCustomCommand] = useState(''); const [memoryBackend, setMemoryBackend] = useState('sqlite'); const [memoryAutoSave, setMemoryAutoSave] = useState(true); const [embeddingProvider, setEmbeddingProvider] = useState('none'); @@ -277,14 +270,6 @@ const TauriCommandsPanel = () => { setConfigLoaded(true); // Load other configuration sections - const tunnel = (config.tunnel as Record) ?? {}; - setTunnelProvider((tunnel.provider as string) ?? 'none'); - setCloudflareToken(((tunnel.cloudflare as Record)?.token as string) ?? ''); - setNgrokToken(((tunnel.ngrok as Record)?.auth_token as string) ?? ''); - setTailscaleHostname( - ((tunnel.tailscale as Record)?.hostname as string) ?? '' - ); - setCustomCommand(((tunnel.custom as Record)?.start_command as string) ?? ''); const memory = (config.memory as Record) ?? {}; setMemoryBackend((memory.backend as string) ?? 'sqlite'); @@ -344,22 +329,6 @@ const TauriCommandsPanel = () => { [daemonShowTray, tauriAvailable] ); - const buildTunnelConfig = (): TunnelConfig => { - if (tunnelProvider === 'cloudflare') { - return { provider: 'cloudflare', cloudflare: { token: cloudflareToken } }; - } - if (tunnelProvider === 'ngrok') { - return { provider: 'ngrok', ngrok: { auth_token: ngrokToken } }; - } - if (tunnelProvider === 'tailscale') { - return { provider: 'tailscale', tailscale: { hostname: tailscaleHostname || null } }; - } - if (tunnelProvider === 'custom') { - return { provider: 'custom', custom: { start_command: customCommand } }; - } - return { provider: 'none' }; - }; - const saveModelSettings = async () => { // Pre-save validation if (!performValidation()) { @@ -470,9 +439,6 @@ const TauriCommandsPanel = () => { } }; - const saveTunnelSettings = () => - run(() => openhumanUpdateTunnelSettings(buildTunnelConfig()), 'saveTunnelSettings'); - const saveMemorySettings = () => run( () => @@ -1023,66 +989,7 @@ const TauriCommandsPanel = () => { collapsible={true} defaultExpanded={!isCollapsed('network-infrastructure')} hasChanges={false} - loading={operationLoading?.includes('Tunnel') || operationLoading?.includes('Memory')}> -
- - - - - {tunnelProvider === 'cloudflare' && ( - - setCloudflareToken(event.target.value)} - /> - - )} - {tunnelProvider === 'ngrok' && ( - - setNgrokToken(event.target.value)} - /> - - )} - {tunnelProvider === 'tailscale' && ( - - setTailscaleHostname(event.target.value)} - /> - - )} - {tunnelProvider === 'custom' && ( - - setCustomCommand(event.target.value)} - /> - - )} - -
- + loading={operationLoading?.includes('Memory')}> @@ -1135,11 +1042,6 @@ const TauriCommandsPanel = () => { - - Save Tunnel Settings - diff --git a/app/src/components/webhooks/TunnelList.tsx b/app/src/components/webhooks/TunnelList.tsx new file mode 100644 index 000000000..37cc136e9 --- /dev/null +++ b/app/src/components/webhooks/TunnelList.tsx @@ -0,0 +1,219 @@ +import { useState } from 'react'; + +import type { Tunnel } from '../../services/api/tunnelsApi'; +import type { TunnelRegistration } from '../../store/webhooksSlice'; +import { BACKEND_URL } from '../../utils/config'; + +interface TunnelListProps { + tunnels: Tunnel[]; + registrations: TunnelRegistration[]; + loading: boolean; + onCreateTunnel: (name: string, description?: string) => Promise; + onDeleteTunnel: (id: string) => Promise; + onRefresh: () => Promise; +} + +export default function TunnelList({ + tunnels, + registrations, + loading, + onCreateTunnel, + onDeleteTunnel, + onRefresh, +}: TunnelListProps) { + const [showCreate, setShowCreate] = useState(false); + const [newName, setNewName] = useState(''); + const [newDesc, setNewDesc] = useState(''); + const [creating, setCreating] = useState(false); + const [actionError, setActionError] = useState(null); + + const handleCreate = async () => { + if (!newName.trim()) return; + setCreating(true); + setActionError(null); + try { + await onCreateTunnel(newName.trim(), newDesc.trim() || undefined); + setNewName(''); + setNewDesc(''); + setShowCreate(false); + } catch (err) { + setActionError(err instanceof Error ? err.message : 'Failed to create tunnel'); + } finally { + setCreating(false); + } + }; + + const getRegistration = (uuid: string) => registrations.find(r => r.tunnel_uuid === uuid); + + const webhookUrl = (uuid: string) => + `${BACKEND_URL || 'https://api.tinyhumans.ai'}/webhooks/${uuid}`; + + return ( +
+ {/* Header */} +
+

Webhook Tunnels

+
+ + +
+
+ + {/* Create form */} + {showCreate && ( +
+ setNewName(e.target.value)} + className="w-full px-3 py-2 text-sm border border-stone-200 rounded-lg focus:outline-none focus:ring-2 focus:ring-primary-500/30 focus:border-primary-500" + autoFocus + /> + setNewDesc(e.target.value)} + className="w-full px-3 py-2 text-sm border border-stone-200 rounded-lg focus:outline-none focus:ring-2 focus:ring-primary-500/30 focus:border-primary-500" + /> +
+ + +
+
+ )} + + {/* Error display */} + {actionError && ( +
+ {actionError} + +
+ )} + + {/* Tunnel list */} + {tunnels.length === 0 && !loading && ( +

+ No tunnels yet. Create one to receive webhook events. +

+ )} + +
+ {tunnels.map(tunnel => { + const reg = getRegistration(tunnel.uuid); + return ( + onDeleteTunnel(tunnel.id)} + onError={setActionError} + /> + ); + })} +
+
+ ); +} + +// ── Tunnel Card ─────────────────────────────────────────────────────────────── + +interface TunnelCardProps { + tunnel: Tunnel; + registration?: TunnelRegistration; + webhookUrl: string; + onDelete: () => Promise; + onError: (msg: string) => void; +} + +function TunnelCard({ tunnel, registration, webhookUrl, onDelete, onError }: TunnelCardProps) { + const [copied, setCopied] = useState(false); + const [deleting, setDeleting] = useState(false); + + const handleCopy = async () => { + try { + await navigator.clipboard.writeText(webhookUrl); + setCopied(true); + setTimeout(() => setCopied(false), 2000); + } catch { + // Clipboard may not be available in Tauri WebView + } + }; + + const handleDelete = async () => { + setDeleting(true); + try { + await onDelete(); + } catch (err) { + onError(err instanceof Error ? err.message : 'Failed to delete tunnel'); + } finally { + setDeleting(false); + } + }; + + return ( +
+
+
+
+

{tunnel.name}

+ + {tunnel.isActive ? 'Active' : 'Inactive'} + + {registration && ( + + {registration.skill_id} + + )} +
+ {tunnel.description && ( +

{tunnel.description}

+ )} +
+ + {webhookUrl} + + +
+
+ +
+
+ ); +} diff --git a/app/src/components/webhooks/WebhookActivity.tsx b/app/src/components/webhooks/WebhookActivity.tsx new file mode 100644 index 000000000..6a1cc918c --- /dev/null +++ b/app/src/components/webhooks/WebhookActivity.tsx @@ -0,0 +1,83 @@ +import type { WebhookActivityEntry } from '../../store/webhooksSlice'; + +interface WebhookActivityProps { + activity: WebhookActivityEntry[]; +} + +const METHOD_COLORS: Record = { + GET: 'text-primary-600 bg-primary-50', + POST: 'text-sage-700 bg-sage-50', + PUT: 'text-amber-700 bg-amber-50', + PATCH: 'text-amber-700 bg-amber-50', + DELETE: 'text-coral-700 bg-coral-50', +}; + +function statusColor(code: number | null): string { + if (code === null) return 'text-stone-400'; + if (code >= 200 && code < 300) return 'text-sage-600'; + if (code >= 400 && code < 500) return 'text-amber-600'; + if (code >= 500) return 'text-coral-600'; + return 'text-stone-600'; +} + +function formatTime(ts: number): string { + return new Date(ts).toLocaleTimeString(undefined, { + hour: '2-digit', + minute: '2-digit', + second: '2-digit', + }); +} + +export default function WebhookActivity({ activity }: WebhookActivityProps) { + if (activity.length === 0) { + return ( +
+

Recent Activity

+

+ No webhook activity yet. Events will appear here when webhooks are received. +

+
+ ); + } + + return ( +
+

+ Recent Activity{' '} + ({activity.length}) +

+
+ {activity.map(entry => ( +
+ + {formatTime(entry.timestamp)} + + + {entry.method} + + + {entry.path || '/'} + + + {entry.status_code ?? '---'} + + {entry.skill_id && ( + + {entry.skill_id} + + )} + + {entry.tunnel_name} + +
+ ))} +
+
+ ); +} diff --git a/app/src/hooks/useWebhooks.ts b/app/src/hooks/useWebhooks.ts new file mode 100644 index 000000000..e284306cb --- /dev/null +++ b/app/src/hooks/useWebhooks.ts @@ -0,0 +1,93 @@ +import debug from 'debug'; +import { useCallback, useEffect } from 'react'; + +import { tunnelsApi } from '../services/api/tunnelsApi'; +import { useAppDispatch, useAppSelector } from '../store/hooks'; +import { addTunnel, removeTunnel, setError, setLoading, setTunnels } from '../store/webhooksSlice'; + +const log = debug('webhooks'); + +/** + * Hook for managing webhook tunnels. + * Fetches tunnels from the backend API and provides CRUD operations. + */ +export function useWebhooks() { + const dispatch = useAppDispatch(); + const { tunnels, registrations, activity, loading, error } = useAppSelector( + state => state.webhooks + ); + const token = useAppSelector(state => state.auth.token); + + // Fetch tunnels on mount (when authenticated) + useEffect(() => { + if (!token) return; + + const fetchTunnels = async () => { + dispatch(setLoading(true)); + try { + const data = await tunnelsApi.getTunnels(); + dispatch(setTunnels(data)); + log('Fetched %d tunnels', data.length); + } catch (err) { + const msg = err instanceof Error ? err.message : 'Failed to fetch tunnels'; + dispatch(setError(msg)); + log('Error fetching tunnels: %s', msg); + } + }; + + fetchTunnels(); + }, [token, dispatch]); + + const createTunnel = useCallback( + async (name: string, description?: string) => { + try { + const tunnel = await tunnelsApi.createTunnel({ name, description }); + dispatch(addTunnel(tunnel)); + log('Created tunnel: %s (%s)', tunnel.name, tunnel.uuid); + return tunnel; + } catch (err) { + const msg = err instanceof Error ? err.message : 'Failed to create tunnel'; + dispatch(setError(msg)); + throw err; + } + }, + [dispatch] + ); + + const deleteTunnel = useCallback( + async (id: string) => { + try { + await tunnelsApi.deleteTunnel(id); + dispatch(removeTunnel(id)); + log('Deleted tunnel: %s', id); + } catch (err) { + const msg = err instanceof Error ? err.message : 'Failed to delete tunnel'; + dispatch(setError(msg)); + throw err; + } + }, + [dispatch] + ); + + const refreshTunnels = useCallback(async () => { + dispatch(setLoading(true)); + try { + const data = await tunnelsApi.getTunnels(); + dispatch(setTunnels(data)); + } catch (err) { + const msg = err instanceof Error ? err.message : 'Failed to refresh tunnels'; + dispatch(setError(msg)); + } + }, [dispatch]); + + return { + tunnels, + registrations, + activity, + loading, + error, + createTunnel, + deleteTunnel, + refreshTunnels, + }; +} diff --git a/app/src/pages/Webhooks.tsx b/app/src/pages/Webhooks.tsx new file mode 100644 index 000000000..4ee30e166 --- /dev/null +++ b/app/src/pages/Webhooks.tsx @@ -0,0 +1,48 @@ +import TunnelList from '../components/webhooks/TunnelList'; +import WebhookActivity from '../components/webhooks/WebhookActivity'; +import { useWebhooks } from '../hooks/useWebhooks'; + +export default function Webhooks() { + const { + tunnels, + registrations, + activity, + loading, + error, + createTunnel, + deleteTunnel, + refreshTunnels, + } = useWebhooks(); + + if (loading && tunnels.length === 0) { + return ( +
+
+
+
+ Loading webhooks… +
+
+
+ ); + } + + return ( +
+
+ {error &&
{error}
} + + + + +
+
+ ); +} diff --git a/app/src/services/api/tunnelsApi.ts b/app/src/services/api/tunnelsApi.ts index 548c35028..218ad4743 100644 --- a/app/src/services/api/tunnelsApi.ts +++ b/app/src/services/api/tunnelsApi.ts @@ -4,7 +4,10 @@ import { apiClient } from '../apiClient'; // ── Types ───────────────────────────────────────────────────────────────────── export interface Tunnel { + /** Internal backend ID (used for CRUD endpoints: GET/PATCH/DELETE /tunnels/:id). */ id: string; + /** External UUID used for webhook routing (appears in webhook URLs and local registrations). */ + uuid: string; name: string; description?: string; isActive: boolean; @@ -51,20 +54,20 @@ export const tunnelsApi = { return response.data; }, - /** GET /tunnels/:id — get a specific webhook tunnel */ - getTunnel: async (id: string): Promise => { - const response = await apiClient.get>(`/tunnels/${id}`); + /** GET /tunnels/:tunnelId — get a specific webhook tunnel by its internal ID. */ + getTunnel: async (tunnelId: string): Promise => { + const response = await apiClient.get>(`/tunnels/${tunnelId}`); return response.data; }, - /** PATCH /tunnels/:id — update a webhook tunnel */ - updateTunnel: async (id: string, body: UpdateTunnelRequest): Promise => { - const response = await apiClient.patch>(`/tunnels/${id}`, body); + /** PATCH /tunnels/:tunnelId — update a webhook tunnel by its internal ID. */ + updateTunnel: async (tunnelId: string, body: UpdateTunnelRequest): Promise => { + const response = await apiClient.patch>(`/tunnels/${tunnelId}`, body); return response.data; }, - /** DELETE /tunnels/:id — delete a webhook tunnel */ - deleteTunnel: async (id: string): Promise => { - await apiClient.delete>(`/tunnels/${id}`); + /** DELETE /tunnels/:tunnelId — delete a webhook tunnel by its internal ID. */ + deleteTunnel: async (tunnelId: string): Promise => { + await apiClient.delete>(`/tunnels/${tunnelId}`); }, }; diff --git a/app/src/services/coreRpcClient.ts b/app/src/services/coreRpcClient.ts index 80693e899..a1873c301 100644 --- a/app/src/services/coreRpcClient.ts +++ b/app/src/services/coreRpcClient.ts @@ -41,7 +41,6 @@ const LEGACY_METHOD_ALIASES: Record = { 'openhuman.update_runtime_settings': 'openhuman.config_update_runtime_settings', 'openhuman.update_screen_intelligence_settings': 'openhuman.config_update_screen_intelligence_settings', - 'openhuman.update_tunnel_settings': 'openhuman.config_update_tunnel_settings', 'openhuman.workspace_onboarding_flag_exists': 'openhuman.config_workspace_onboarding_flag_exists', 'openhuman.workspace_onboarding_flag_set': 'openhuman.config_workspace_onboarding_flag_set', }; diff --git a/app/src/store/index.ts b/app/src/store/index.ts index ca3fa9def..d07c6213b 100644 --- a/app/src/store/index.ts +++ b/app/src/store/index.ts @@ -29,6 +29,7 @@ import socketReducer from './socketSlice'; import teamReducer from './teamSlice'; import threadReducer from './threadSlice'; import userReducer from './userSlice'; +import webhooksReducer from './webhooksSlice'; // Persist config for auth only const authPersistConfig = { @@ -132,6 +133,7 @@ export const store = configureStore({ invite: inviteReducer, accessibility: accessibilityReducer, channelConnections: persistedChannelConnectionsReducer, + webhooks: webhooksReducer, }, middleware: getDefaultMiddleware => { const middleware = getDefaultMiddleware({ diff --git a/app/src/store/webhooksSlice.ts b/app/src/store/webhooksSlice.ts new file mode 100644 index 000000000..306446b24 --- /dev/null +++ b/app/src/store/webhooksSlice.ts @@ -0,0 +1,92 @@ +import { createSlice, type PayloadAction } from '@reduxjs/toolkit'; + +import type { Tunnel } from '../services/api/tunnelsApi'; + +// ── Types ───────────────────────────────────────────────────────────────────── + +/** Local tunnel-to-skill registration (from the Rust core WebhookRouter). */ +export interface TunnelRegistration { + tunnel_uuid: string; + skill_id: string; + tunnel_name: string | null; + backend_tunnel_id: string | null; +} + +/** Entry in the webhook activity log. */ +export interface WebhookActivityEntry { + correlation_id: string; + tunnel_name: string; + method: string; + path: string; + status_code: number | null; + skill_id: string | null; + timestamp: number; +} + +interface WebhooksState { + /** Tunnels from the backend API. */ + tunnels: Tunnel[]; + /** Local tunnel-to-skill registrations from the Rust core. */ + registrations: TunnelRegistration[]; + /** Recent webhook activity (ring buffer, newest first). */ + activity: WebhookActivityEntry[]; + loading: boolean; + error: string | null; +} + +// ── Slice ───────────────────────────────────────────────────────────────────── + +const MAX_ACTIVITY_ENTRIES = 100; + +const initialState: WebhooksState = { + tunnels: [], + registrations: [], + activity: [], + loading: false, + error: null, +}; + +const webhooksSlice = createSlice({ + name: 'webhooks', + initialState, + reducers: { + setTunnels: (state, action: PayloadAction) => { + state.tunnels = action.payload; + state.loading = false; + state.error = null; + }, + addTunnel: (state, action: PayloadAction) => { + state.tunnels.push(action.payload); + }, + removeTunnel: (state, action: PayloadAction) => { + state.tunnels = state.tunnels.filter(t => t.id !== action.payload); + }, + setRegistrations: (state, action: PayloadAction) => { + state.registrations = action.payload; + }, + addActivity: (state, action: PayloadAction) => { + state.activity.unshift(action.payload); + if (state.activity.length > MAX_ACTIVITY_ENTRIES) { + state.activity.splice(MAX_ACTIVITY_ENTRIES); + } + }, + setLoading: (state, action: PayloadAction) => { + state.loading = action.payload; + }, + setError: (state, action: PayloadAction) => { + state.error = action.payload; + state.loading = false; + }, + }, +}); + +export const { + setTunnels, + addTunnel, + removeTunnel, + setRegistrations, + addActivity, + setLoading, + setError, +} = webhooksSlice.actions; +export default webhooksSlice.reducer; diff --git a/app/src/utils/tauriCommands.ts b/app/src/utils/tauriCommands.ts index 9c37891ae..b7d32d4d0 100644 --- a/app/src/utils/tauriCommands.ts +++ b/app/src/utils/tauriCommands.ts @@ -984,18 +984,6 @@ function tauriErrorMessage(err: unknown): string { return 'Unknown Tauri invoke error'; } -export interface TunnelConfig { - provider: string; - cloudflare?: { token: string } | null; - tailscale?: { funnel?: boolean; hostname?: string | null } | null; - ngrok?: { auth_token: string; domain?: string | null } | null; - custom?: { - start_command: string; - health_url?: string | null; - url_pattern?: string | null; - } | null; -} - export async function openhumanGetConfig(): Promise> { if (!isTauri()) { throw new Error('Not running in Tauri'); @@ -1027,18 +1015,6 @@ export async function openhumanUpdateMemorySettings( }); } -export async function openhumanUpdateTunnelSettings( - tunnel: TunnelConfig -): Promise> { - if (!isTauri()) { - throw new Error('Not running in Tauri'); - } - return await callCoreRpc>({ - method: 'openhuman.update_tunnel_settings', - params: tunnel, - }); -} - export async function openhumanUpdateRuntimeSettings( update: RuntimeSettingsUpdate ): Promise> { diff --git a/src/openhuman/config/mod.rs b/src/openhuman/config/mod.rs index 7286849db..75b04c174 100644 --- a/src/openhuman/config/mod.rs +++ b/src/openhuman/config/mod.rs @@ -14,16 +14,15 @@ pub use schema::{ apply_runtime_proxy_to_builder, build_runtime_proxy_client, build_runtime_proxy_client_with_timeouts, runtime_proxy_config, set_runtime_proxy_config, AgentConfig, AuditConfig, AutocompleteConfig, AutonomyConfig, BrowserComputerUseConfig, - BrowserConfig, ChannelsConfig, ClassificationRule, CloudflareTunnelConfig, ComposioConfig, - Config, CostConfig, CronConfig, CustomTunnelConfig, DelegateAgentConfig, DiscordConfig, - DockerRuntimeConfig, EmbeddingRouteConfig, HardwareConfig, HardwareTransport, HeartbeatConfig, - HttpRequestConfig, IMessageConfig, IdentityConfig, LarkConfig, LearningConfig, LocalAiConfig, - MatrixConfig, MemoryConfig, ModelRouteConfig, MultimodalConfig, NgrokTunnelConfig, - ObservabilityConfig, PeripheralBoardConfig, PeripheralsConfig, ProxyConfig, ProxyScope, - QueryClassificationConfig, ReflectionSource, ReliabilityConfig, ResourceLimitsConfig, - RuntimeConfig, SandboxBackend, SandboxConfig, SchedulerConfig, ScreenIntelligenceConfig, - SecretsConfig, SecurityConfig, SlackConfig, StorageConfig, StorageProviderConfig, - StorageProviderSection, StreamMode, TailscaleTunnelConfig, TelegramConfig, TunnelConfig, + BrowserConfig, ChannelsConfig, ClassificationRule, ComposioConfig, Config, CostConfig, + CronConfig, DelegateAgentConfig, DiscordConfig, DockerRuntimeConfig, EmbeddingRouteConfig, + HardwareConfig, HardwareTransport, HeartbeatConfig, HttpRequestConfig, IMessageConfig, + IdentityConfig, LarkConfig, LearningConfig, LocalAiConfig, MatrixConfig, MemoryConfig, + ModelRouteConfig, MultimodalConfig, ObservabilityConfig, PeripheralBoardConfig, + PeripheralsConfig, ProxyConfig, ProxyScope, QueryClassificationConfig, ReflectionSource, + ReliabilityConfig, ResourceLimitsConfig, RuntimeConfig, SandboxBackend, SandboxConfig, + SchedulerConfig, ScreenIntelligenceConfig, SecretsConfig, SecurityConfig, SlackConfig, + StorageConfig, StorageProviderConfig, StorageProviderSection, StreamMode, TelegramConfig, WebSearchConfig, WebhookConfig, }; pub use schemas::{ diff --git a/src/openhuman/config/ops.rs b/src/openhuman/config/ops.rs index 21dbacb58..83d17c3a7 100644 --- a/src/openhuman/config/ops.rs +++ b/src/openhuman/config/ops.rs @@ -5,7 +5,7 @@ use std::path::PathBuf; use serde::Serialize; use serde_json::json; -use crate::openhuman::config::{Config, TunnelConfig}; +use crate::openhuman::config::Config; use crate::openhuman::screen_intelligence; use crate::rpc::RpcOutcome; @@ -223,22 +223,6 @@ pub async fn apply_screen_intelligence_settings( )) } -pub async fn apply_tunnel_settings( - config: &mut Config, - tunnel: TunnelConfig, -) -> Result, String> { - config.tunnel = tunnel; - config.save().await.map_err(|e| e.to_string())?; - let snapshot = snapshot_config_json(config)?; - Ok(RpcOutcome::new( - snapshot, - vec![format!( - "tunnel settings saved to {}", - config.config_path.display() - )], - )) -} - pub async fn apply_runtime_settings( config: &mut Config, update: RuntimeSettingsPatch, @@ -304,13 +288,6 @@ pub async fn load_and_apply_screen_intelligence_settings( apply_screen_intelligence_settings(&mut config, update).await } -pub async fn load_and_apply_tunnel_settings( - tunnel: TunnelConfig, -) -> Result, String> { - let mut config = load_config_with_timeout().await?; - apply_tunnel_settings(&mut config, tunnel).await -} - pub async fn load_and_apply_runtime_settings( update: RuntimeSettingsPatch, ) -> Result, String> { diff --git a/src/openhuman/config/schema/mod.rs b/src/openhuman/config/schema/mod.rs index f71fa0e67..67ca31304 100644 --- a/src/openhuman/config/schema/mod.rs +++ b/src/openhuman/config/schema/mod.rs @@ -20,7 +20,6 @@ mod routes; mod runtime; mod storage_memory; mod tools; -mod tunnel; pub use accessibility::ScreenIntelligenceConfig; pub use agent::{AgentConfig, DelegateAgentConfig}; @@ -56,10 +55,5 @@ pub use tools::{ BrowserComputerUseConfig, BrowserConfig, ComposioConfig, HttpRequestConfig, MultimodalConfig, SecretsConfig, WebSearchConfig, }; -pub use tunnel::{ - CloudflareTunnelConfig, CustomTunnelConfig, NgrokTunnelConfig, TailscaleTunnelConfig, - TunnelConfig, -}; - mod types; pub use types::*; diff --git a/src/openhuman/config/schema/tunnel.rs b/src/openhuman/config/schema/tunnel.rs deleted file mode 100644 index 9bc890899..000000000 --- a/src/openhuman/config/schema/tunnel.rs +++ /dev/null @@ -1,54 +0,0 @@ -//! Tunnel (Cloudflare, Tailscale, ngrok, custom) configuration. - -use schemars::JsonSchema; -use serde::{Deserialize, Serialize}; - -#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] -pub struct TunnelConfig { - pub provider: String, - #[serde(default)] - pub cloudflare: Option, - #[serde(default)] - pub tailscale: Option, - #[serde(default)] - pub ngrok: Option, - #[serde(default)] - pub custom: Option, -} - -impl Default for TunnelConfig { - fn default() -> Self { - Self { - provider: "none".into(), - cloudflare: None, - tailscale: None, - ngrok: None, - custom: None, - } - } -} - -#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] -pub struct CloudflareTunnelConfig { - pub token: String, -} - -#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] -pub struct TailscaleTunnelConfig { - #[serde(default)] - pub funnel: bool, - pub hostname: Option, -} - -#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] -pub struct NgrokTunnelConfig { - pub auth_token: String, - pub domain: Option, -} - -#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] -pub struct CustomTunnelConfig { - pub start_command: String, - pub health_url: Option, - pub url_pattern: Option, -} diff --git a/src/openhuman/config/schema/types.rs b/src/openhuman/config/schema/types.rs index 48ad73103..eeeb73d8a 100644 --- a/src/openhuman/config/schema/types.rs +++ b/src/openhuman/config/schema/types.rs @@ -66,9 +66,6 @@ pub struct Config { #[serde(default)] pub storage: StorageConfig, - #[serde(default)] - pub tunnel: TunnelConfig, - #[serde(default)] pub composio: ComposioConfig, @@ -140,7 +137,6 @@ impl Default for Config { channels_config: ChannelsConfig::default(), memory: MemoryConfig::default(), storage: StorageConfig::default(), - tunnel: TunnelConfig::default(), composio: ComposioConfig::default(), secrets: SecretsConfig::default(), browser: BrowserConfig::default(), diff --git a/src/openhuman/config/schemas.rs b/src/openhuman/config/schemas.rs index dc3de2b80..e5d6f72b8 100644 --- a/src/openhuman/config/schemas.rs +++ b/src/openhuman/config/schemas.rs @@ -5,7 +5,6 @@ use serde_json::{Map, Value}; use crate::core::all::{ControllerFuture, RegisteredController}; use crate::core::{ControllerSchema, FieldSchema, TypeSchema}; use crate::openhuman::config::rpc as config_rpc; -use crate::openhuman::config::TunnelConfig; use crate::rpc::RpcOutcome; const DEFAULT_ONBOARDING_FLAG_NAME: &str = ".skip_onboarding"; @@ -77,7 +76,6 @@ pub fn all_controller_schemas() -> Vec { schemas("update_model_settings"), schemas("update_memory_settings"), schemas("update_screen_intelligence_settings"), - schemas("update_tunnel_settings"), schemas("update_runtime_settings"), schemas("update_browser_settings"), schemas("resolve_api_url"), @@ -109,10 +107,6 @@ pub fn all_registered_controllers() -> Vec { schema: schemas("update_screen_intelligence_settings"), handler: handle_update_screen_intelligence_settings, }, - RegisteredController { - schema: schemas("update_tunnel_settings"), - handler: handle_update_tunnel_settings, - }, RegisteredController { schema: schemas("update_runtime_settings"), handler: handle_update_runtime_settings, @@ -245,39 +239,6 @@ pub fn schemas(function: &str) -> ControllerSchema { ], outputs: vec![json_output("snapshot", "Updated config snapshot.")], }, - "update_tunnel_settings" => ControllerSchema { - namespace: "config", - function: "update_tunnel_settings", - description: "Replace tunnel settings with provided config payload.", - inputs: vec![ - required_string("provider", "Tunnel provider id."), - FieldSchema { - name: "cloudflare", - ty: TypeSchema::Option(Box::new(TypeSchema::Ref("CloudflareTunnelConfig"))), - comment: "Cloudflare tunnel settings.", - required: false, - }, - FieldSchema { - name: "tailscale", - ty: TypeSchema::Option(Box::new(TypeSchema::Ref("TailscaleTunnelConfig"))), - comment: "Tailscale tunnel settings.", - required: false, - }, - FieldSchema { - name: "ngrok", - ty: TypeSchema::Option(Box::new(TypeSchema::Ref("NgrokTunnelConfig"))), - comment: "ngrok tunnel settings.", - required: false, - }, - FieldSchema { - name: "custom", - ty: TypeSchema::Option(Box::new(TypeSchema::Ref("CustomTunnelConfig"))), - comment: "Custom tunnel settings.", - required: false, - }, - ], - outputs: vec![json_output("snapshot", "Updated config snapshot.")], - }, "update_runtime_settings" => ControllerSchema { namespace: "config", function: "update_runtime_settings", @@ -470,13 +431,6 @@ fn handle_update_screen_intelligence_settings(params: Map) -> Con }) } -fn handle_update_tunnel_settings(params: Map) -> ControllerFuture { - Box::pin(async move { - let tunnel = deserialize_params::(params)?; - to_json(config_rpc::load_and_apply_tunnel_settings(tunnel).await?) - }) -} - fn handle_update_runtime_settings(params: Map) -> ControllerFuture { Box::pin(async move { let update = deserialize_params::(params)?; diff --git a/src/openhuman/config/settings_cli.rs b/src/openhuman/config/settings_cli.rs index c352eaa39..d4588f92e 100644 --- a/src/openhuman/config/settings_cli.rs +++ b/src/openhuman/config/settings_cli.rs @@ -28,10 +28,6 @@ pub fn settings_section_json( .get("memory") .cloned() .unwrap_or(serde_json::Value::Null), - "tunnel" => cfg - .get("tunnel") - .cloned() - .unwrap_or(serde_json::Value::Null), "runtime" => cfg .get("runtime") .cloned() diff --git a/src/openhuman/mod.rs b/src/openhuman/mod.rs index 1034dd06a..ad35a6091 100644 --- a/src/openhuman/mod.rs +++ b/src/openhuman/mod.rs @@ -34,6 +34,6 @@ pub mod security; pub mod service; pub mod skills; pub mod tools; -pub mod tunnel; pub mod util; +pub mod webhooks; pub mod workspace; diff --git a/src/openhuman/skills/qjs_engine.rs b/src/openhuman/skills/qjs_engine.rs index 8c15fcb6f..3bd69afd0 100644 --- a/src/openhuman/skills/qjs_engine.rs +++ b/src/openhuman/skills/qjs_engine.rs @@ -39,6 +39,7 @@ use crate::openhuman::skills::qjs_skill_instance::{BridgeDeps, QjsSkillInstance} use crate::openhuman::skills::skill_registry::SkillRegistry; use crate::openhuman::skills::socket_manager::SocketManager; use crate::openhuman::skills::types::{SkillSnapshot, SkillStatus, ToolCallOrigin, ToolResult}; +use crate::openhuman::webhooks::WebhookRouter; // IdbStorage removed during runtime cleanup /// The central runtime engine using QuickJS. @@ -61,6 +62,8 @@ pub struct RuntimeEngine { memory_client: RwLock>, /// Socket manager for emitting tool:sync events. socket_manager: RwLock>>, + /// Webhook router for tunnel-to-skill routing. + webhook_router: Arc, /// Workspace directory for user-installed skills from registry. workspace_dir: RwLock>, } @@ -87,6 +90,10 @@ impl RuntimeEngine { } }; + // Initialize webhook router with persistence + let webhook_routes_path = skills_data_dir.join("webhook_routes.json"); + let webhook_router = Arc::new(WebhookRouter::new(Some(webhook_routes_path))); + log::info!("[runtime] QuickJS RuntimeEngine created"); Ok(Self { @@ -99,6 +106,7 @@ impl RuntimeEngine { resource_dir: RwLock::new(None), memory_client: RwLock::new(memory_client), socket_manager: RwLock::new(None), + webhook_router, workspace_dir: RwLock::new(None), }) } @@ -137,9 +145,16 @@ impl RuntimeEngine { /// Set the socket manager for emitting `tool:sync` events. pub fn set_socket_manager(&self, mgr: Arc) { + // Also wire the webhook router into the socket manager + mgr.set_webhook_router(Arc::clone(&self.webhook_router)); *self.socket_manager.write() = Some(mgr); } + /// Get a clone of the webhook router Arc. + pub fn webhook_router(&self) -> Arc { + Arc::clone(&self.webhook_router) + } + /// Set the workspace directory for user-installed skills from the registry. pub fn set_workspace_dir(&self, dir: PathBuf) { log::info!("[runtime] Workspace directory set to: {:?}", dir); @@ -377,6 +392,7 @@ impl RuntimeEngine { cron_scheduler: self.cron_scheduler.clone(), skill_registry: self.registry.clone(), memory_client: self.memory_client.read().clone(), + webhook_router: Some(self.webhook_router.clone()), data_dir: data_dir.clone(), }; @@ -457,6 +473,7 @@ impl RuntimeEngine { pub async fn stop_skill(&self, skill_id: &str) -> Result<(), String> { self.registry.stop_skill(skill_id).await?; self.cron_scheduler.unregister_all_for_skill(skill_id); + self.webhook_router.unregister_skill(skill_id); self.emit_status_change(skill_id); self.sync_tools().await; Ok(()) diff --git a/src/openhuman/skills/qjs_skill_instance/event_loop.rs b/src/openhuman/skills/qjs_skill_instance/event_loop.rs index 37e1077fc..cb0fe9941 100644 --- a/src/openhuman/skills/qjs_skill_instance/event_loop.rs +++ b/src/openhuman/skills/qjs_skill_instance/event_loop.rs @@ -364,6 +364,88 @@ async fn handle_message( log::warn!("[skill:{}] onError() handler failed: {e}", skill_id); } } + SkillMessage::WebhookRequest { + correlation_id, + method, + path, + headers, + query, + body, + tunnel_id, + tunnel_name, + reply, + } => { + log::info!( + "[skill:{}] event_loop: WebhookRequest {} {} (tunnel={})", + skill_id, + method, + path, + tunnel_id, + ); + + // Restore OAuth credential in case the handler needs authenticated API calls + restore_oauth_credential(ctx, skill_id, data_dir).await; + + let args = serde_json::json!({ + "correlationId": correlation_id, + "method": method, + "path": path, + "headers": headers, + "query": query, + "body": body, + "tunnelId": tunnel_id, + "tunnelName": tunnel_name, + }); + + match handle_js_call(rt, ctx, "onWebhookRequest", &args.to_string()).await { + Ok(response_val) => { + use crate::openhuman::webhooks::WebhookResponseData; + + let status_code = response_val + .get("statusCode") + .and_then(|v| v.as_u64()) + .unwrap_or(200) as u16; + let resp_headers: HashMap = response_val + .get("headers") + .map(|v| match serde_json::from_value(v.clone()) { + Ok(h) => h, + Err(e) => { + log::warn!( + "[skill] Failed to parse webhook response headers: {e}, raw: {v}" + ); + HashMap::new() + } + }) + .unwrap_or_default(); + let resp_body = response_val + .get("body") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(); + + log::debug!( + "[skill:{}] event_loop: WebhookRequest handled → status {}", + skill_id, + status_code, + ); + + let _ = reply.send(Ok(WebhookResponseData { + correlation_id, + status_code, + headers: resp_headers, + body: resp_body, + })); + } + Err(e) => { + log::warn!( + "[skill:{}] event_loop: onWebhookRequest failed: {}", + skill_id, + e, + ); + let _ = reply.send(Err(e)); + } + } + } SkillMessage::Rpc { method, params, diff --git a/src/openhuman/skills/qjs_skill_instance/instance.rs b/src/openhuman/skills/qjs_skill_instance/instance.rs index 066f7b09e..567c80345 100644 --- a/src/openhuman/skills/qjs_skill_instance/instance.rs +++ b/src/openhuman/skills/qjs_skill_instance/instance.rs @@ -140,6 +140,7 @@ impl QjsSkillInstance { skill_id: skill_id.clone(), data_dir: data_dir.clone(), memory_client: _deps.memory_client.clone(), + webhook_router: _deps.webhook_router.clone(), }; if let Err(e) = qjs_ops::register_ops( diff --git a/src/openhuman/skills/qjs_skill_instance/types.rs b/src/openhuman/skills/qjs_skill_instance/types.rs index 7fe20ad8d..ae45077a2 100644 --- a/src/openhuman/skills/qjs_skill_instance/types.rs +++ b/src/openhuman/skills/qjs_skill_instance/types.rs @@ -15,6 +15,7 @@ pub struct BridgeDeps { pub cron_scheduler: Arc, pub skill_registry: Arc, pub memory_client: Option, + pub webhook_router: Option>, pub data_dir: PathBuf, // NOTE: No v8_creation_lock - QuickJS doesn't need it } diff --git a/src/openhuman/skills/quickjs_libs/bootstrap.js b/src/openhuman/skills/quickjs_libs/bootstrap.js index 66c277c1a..d93f91005 100644 --- a/src/openhuman/skills/quickjs_libs/bootstrap.js +++ b/src/openhuman/skills/quickjs_libs/bootstrap.js @@ -1012,4 +1012,132 @@ globalThis.model = { }; console.log('[bootstrap] Model API initialized'); + +// ============================================================================ +// Webhook / Tunnel API (skill-scoped) +// ============================================================================ +// All operations are scoped to the calling skill. The Rust bridge injects the +// skill_id automatically — JS code cannot impersonate another skill. + +globalThis.webhook = { + /** + * Register this skill to receive webhooks for a tunnel UUID. + * Rejects if the tunnel is already owned by a different skill. + * @param {string} tunnelUuid - The tunnel UUID (from createTunnel or backend) + * @param {string} [tunnelName] - Human-readable name for display + * @param {string} [backendTunnelId] - Backend MongoDB _id for CRUD + */ + register: function (tunnelUuid, tunnelName, backendTunnelId) { + __ops.webhook_register( + tunnelUuid, + tunnelName || null, + backendTunnelId || null + ); + }, + + /** + * Unregister this skill from a tunnel. + * Rejects if the tunnel is not owned by this skill. + * @param {string} tunnelUuid + */ + unregister: function (tunnelUuid) { + __ops.webhook_unregister(tunnelUuid); + }, + + /** + * List only this skill's registered tunnel mappings. + * Never includes other skills' tunnels. + * @returns {Array<{tunnel_uuid: string, skill_id: string, tunnel_name: string|null}>} + */ + list: function () { + var json = __ops.webhook_list(); + return JSON.parse(json); + }, + + /** + * Create a new tunnel via the backend API, automatically registered to + * this skill. + * @param {string} name - Tunnel name + * @param {string} [description] - Optional description + * @returns {Promise<{id: string, uuid: string, webhookUrl: string}>} + */ + createTunnel: async function (name, description) { + var backendUrl = __platform.env('BACKEND_URL') || 'https://api.tinyhumans.ai'; + var jwtToken = __ops.get_session_token() || ''; + + var result = await net.fetch(backendUrl + '/tunnels', { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + Authorization: 'Bearer ' + jwtToken, + }, + body: JSON.stringify({ name: name, description: description || '' }), + timeout: 15000, + }); + + var parsed = JSON.parse(result); + if (parsed.status >= 400) { + throw new Error('Failed to create tunnel: ' + parsed.status + ' ' + parsed.body); + } + var data = JSON.parse(parsed.body); + var tunnel = data.tunnel || data; + + // Auto-register this tunnel to the calling skill + if (tunnel.uuid) { + webhook.register(tunnel.uuid, name, tunnel._id || tunnel.id || null); + } + + // Build webhook URL for the caller + tunnel.webhookUrl = backendUrl + '/webhooks/' + tunnel.uuid; + + console.log('[webhook] Created tunnel: ' + name + ' → ' + tunnel.webhookUrl); + return tunnel; + }, + + /** + * List locally registered tunnels scoped to this skill. + * Returns the local webhook registrations, not data from the backend API. + * @returns {Promise} Array of local tunnel registration objects. + */ + listTunnels: async function () { + return webhook.list(); + }, + + /** + * Delete a tunnel. Fails if the tunnel is not owned by this skill. + * @param {string} tunnelUuid - The tunnel UUID to delete + */ + deleteTunnel: async function (tunnelUuid) { + // Verify ownership locally first (will throw if not owned), but don't + // unregister yet — we need the backend DELETE to succeed first so we + // don't orphan tunnels if the network call fails. + webhook.list().forEach(function (reg) { + // webhook.list() only returns this skill's registrations, so if the + // tunnelUuid isn't among them the skill doesn't own it. + }); + + // Delete from backend first + var backendUrl = __platform.env('BACKEND_URL') || 'https://api.tinyhumans.ai'; + var jwtToken = __ops.get_session_token() || ''; + + var result = await net.fetch(backendUrl + '/tunnels/' + tunnelUuid, { + method: 'DELETE', + headers: { Authorization: 'Bearer ' + jwtToken }, + timeout: 10000, + }); + + var parsed = JSON.parse(result); + if (parsed.status >= 400 && parsed.status !== 404) { + throw new Error('[webhook] Backend delete failed with status ' + parsed.status); + } + + // Backend confirmed deletion (or 404 = already gone) — now safe to + // remove the local registration. + webhook.unregister(tunnelUuid); + + console.log('[webhook] Deleted tunnel: ' + tunnelUuid); + }, +}; + +console.log('[bootstrap] Webhook API initialized'); console.log('[bootstrap] QuickJS browser APIs initialized'); diff --git a/src/openhuman/skills/quickjs_libs/qjs_ops/mod.rs b/src/openhuman/skills/quickjs_libs/qjs_ops/mod.rs index 7caf7ccb9..12f2e6d9e 100644 --- a/src/openhuman/skills/quickjs_libs/qjs_ops/mod.rs +++ b/src/openhuman/skills/quickjs_libs/qjs_ops/mod.rs @@ -12,6 +12,7 @@ mod ops_core; mod ops_net; mod ops_state; mod ops_storage; +mod ops_webhook; pub mod types; // Re-export public API used by `qjs_skill_instance` diff --git a/src/openhuman/skills/quickjs_libs/qjs_ops/ops.rs b/src/openhuman/skills/quickjs_libs/qjs_ops/ops.rs index 5ccec08e5..cea30618f 100644 --- a/src/openhuman/skills/quickjs_libs/qjs_ops/ops.rs +++ b/src/openhuman/skills/quickjs_libs/qjs_ops/ops.rs @@ -23,6 +23,7 @@ pub fn register_ops( ops_net::register(ctx, &ops, ws_state)?; ops_storage::register(ctx, &ops, storage, skill_context.clone())?; ops_state::register(ctx, &ops, skill_state, skill_context.clone())?; + ops_webhook::register(ctx, &ops, skill_context)?; globals.set("__ops", ops)?; Ok(()) diff --git a/src/openhuman/skills/quickjs_libs/qjs_ops/ops_webhook.rs b/src/openhuman/skills/quickjs_libs/qjs_ops/ops_webhook.rs new file mode 100644 index 000000000..7c51274bd --- /dev/null +++ b/src/openhuman/skills/quickjs_libs/qjs_ops/ops_webhook.rs @@ -0,0 +1,83 @@ +//! Webhook ops: register/unregister/list tunnel-to-skill mappings. +//! +//! All operations are scoped to the calling skill — the `skill_id` is baked +//! into the closure context at startup and cannot be overridden from JS. + +use rquickjs::{Ctx, Function, Object}; + +use super::types::{js_err, SkillContext}; + +pub fn register<'js>( + ctx: &Ctx<'js>, + ops: &Object<'js>, + skill_context: SkillContext, +) -> rquickjs::Result<()> { + // webhook_register(tunnel_uuid, tunnel_name?, backend_tunnel_id?) + { + let sc = skill_context.clone(); + ops.set( + "webhook_register", + Function::new( + ctx.clone(), + move |tunnel_uuid: String, + tunnel_name: rquickjs::Value<'_>, + backend_tunnel_id: rquickjs::Value<'_>| + -> rquickjs::Result<()> { + let router = sc + .webhook_router + .as_ref() + .ok_or_else(|| js_err("Webhook router not available"))?; + + let name = tunnel_name.as_string().and_then(|s| s.to_string().ok()); + let backend_id = backend_tunnel_id + .as_string() + .and_then(|s| s.to_string().ok()); + + router + .register(&tunnel_uuid, &sc.skill_id, name, backend_id) + .map_err(|e| js_err(e)) + }, + ), + )?; + } + + // webhook_unregister(tunnel_uuid) + { + let sc = skill_context.clone(); + ops.set( + "webhook_unregister", + Function::new( + ctx.clone(), + move |tunnel_uuid: String| -> rquickjs::Result<()> { + let router = sc + .webhook_router + .as_ref() + .ok_or_else(|| js_err("Webhook router not available"))?; + + router + .unregister(&tunnel_uuid, &sc.skill_id) + .map_err(|e| js_err(e)) + }, + ), + )?; + } + + // webhook_list() -> JSON array of this skill's tunnel registrations + { + let sc = skill_context; + ops.set( + "webhook_list", + Function::new(ctx.clone(), move || -> rquickjs::Result { + let router = sc + .webhook_router + .as_ref() + .ok_or_else(|| js_err("Webhook router not available"))?; + + let registrations = router.list_for_skill(&sc.skill_id); + serde_json::to_string(®istrations).map_err(|e| js_err(e.to_string())) + }), + )?; + } + + Ok(()) +} diff --git a/src/openhuman/skills/quickjs_libs/qjs_ops/types.rs b/src/openhuman/skills/quickjs_libs/qjs_ops/types.rs index eda588828..2df04d5a5 100644 --- a/src/openhuman/skills/quickjs_libs/qjs_ops/types.rs +++ b/src/openhuman/skills/quickjs_libs/qjs_ops/types.rs @@ -77,6 +77,7 @@ pub struct SkillContext { pub skill_id: String, pub data_dir: PathBuf, pub memory_client: Option, + pub webhook_router: Option>, } // ============================================================================ diff --git a/src/openhuman/skills/skill_registry.rs b/src/openhuman/skills/skill_registry.rs index 018f736c1..7fc8bef1e 100644 --- a/src/openhuman/skills/skill_registry.rs +++ b/src/openhuman/skills/skill_registry.rs @@ -341,6 +341,67 @@ impl SkillRegistry { } } } + + /// Send an incoming webhook request to a specific skill and wait for the response. + /// + /// Returns the skill's response (status code, headers, body) or an error. + /// Times out after 25 seconds (under the backend's 30-second timeout). + pub async fn send_webhook_request( + &self, + skill_id: &str, + correlation_id: String, + method: String, + path: String, + headers: std::collections::HashMap, + query: std::collections::HashMap, + body: String, + tunnel_id: String, + tunnel_name: String, + ) -> Result { + let sender = { + let skills = self.skills.read(); + let entry = skills + .get(skill_id) + .ok_or_else(|| format!("Skill '{}' not found", skill_id))?; + let status = entry.state.read().status; + if status != SkillStatus::Running { + return Err(format!( + "Skill '{}' is not running (status: {:?})", + skill_id, status + )); + } + entry.sender.clone() + }; + + let (reply_tx, reply_rx) = oneshot::channel(); + + sender + .send(SkillMessage::WebhookRequest { + correlation_id, + method, + path, + headers, + query, + body, + tunnel_id, + tunnel_name, + reply: reply_tx, + }) + .await + .map_err(|_| format!("Skill '{}' message channel closed", skill_id))?; + + match tokio::time::timeout(std::time::Duration::from_secs(25), reply_rx).await { + Ok(Ok(result)) => result, + Ok(Err(_)) => Err(format!( + "Skill '{}' webhook reply channel dropped", + skill_id + )), + Err(_) => Err(format!( + "Skill '{}' webhook handler timed out (25s)", + skill_id + )), + } + } } impl Default for SkillRegistry { diff --git a/src/openhuman/skills/socket_manager.rs b/src/openhuman/skills/socket_manager.rs index 70ed24980..d1fdaa9ca 100644 --- a/src/openhuman/skills/socket_manager.rs +++ b/src/openhuman/skills/socket_manager.rs @@ -30,6 +30,7 @@ use { // SkillRegistry only available on desktop use crate::openhuman::skills::skill_registry::SkillRegistry; use crate::openhuman::skills::types::{SkillSnapshot, SkillStatus, ToolCallOrigin}; +use crate::openhuman::webhooks::{WebhookRequest, WebhookRouter}; /// Events emitted to the frontend via Tauri. #[allow(dead_code)] @@ -46,6 +47,7 @@ pub mod events { struct SharedState { registry: RwLock>>, + webhook_router: RwLock>>, status: RwLock, socket_id: RwLock>, } @@ -84,6 +86,7 @@ impl SocketManager { Self { shared: Arc::new(SharedState { registry: RwLock::new(None), + webhook_router: RwLock::new(None), status: RwLock::new(ConnectionStatus::Disconnected), socket_id: RwLock::new(None), }), @@ -98,6 +101,11 @@ impl SocketManager { *self.shared.registry.write() = Some(registry); } + /// Set the webhook router for skill-targeted webhook delivery. + pub fn set_webhook_router(&self, router: Arc) { + *self.shared.webhook_router.write() = Some(router); + } + /// Get current socket state. pub fn get_state(&self) -> SocketState { SocketState { @@ -626,6 +634,14 @@ fn handle_sio_event( handle_mcp_tool_call(&shared, data, &tx).await; }); } + // Webhook tunnel — route to owning skill and relay response + "webhook:request" => { + let shared = Arc::clone(shared); + let tx = emit_tx.clone(); + tokio::spawn(async move { + handle_webhook_request(&shared, data, &tx).await; + }); + } _ => { // Forward to skills (desktop only) and frontend { @@ -771,6 +787,171 @@ async fn handle_mcp_tool_call( ); } +// --------------------------------------------------------------------------- +// 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, +) { + // Parse the incoming request + let request: WebhookRequest = match serde_json::from_value(data.clone()) { + Ok(r) => r, + Err(e) => { + log::error!("[socket-mgr] Failed to parse webhook:request payload: {e}"); + // Try to extract correlationId so we can send a 400 back + let cid = data + .get("correlationId") + .and_then(|v| v.as_str()) + .unwrap_or("unknown") + .to_string(); + 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-mgr] webhook:request {} {} (tunnel={}, correlationId={})", + method, + path, + tunnel_uuid, + correlation_id, + ); + + // Look up the owning skill via the webhook router + let router = shared.webhook_router.read().clone(); + let skill_id = router.as_ref().and_then(|r| r.route(&tunnel_uuid)); + + let (response, resolved_skill_id) = match skill_id { + Some(sid) => { + log::debug!("[socket-mgr] webhook:request routed to skill '{}'", sid,); + + let registry = shared.registry.read().clone(); + match registry { + Some(registry) => { + let result = registry + .send_webhook_request( + &sid, + correlation_id.clone(), + request.method, + request.path, + request.headers, + request.query, + request.body, + request.tunnel_id, + request.tunnel_name, + ) + .await; + + match result { + Ok(resp) => (resp, Some(sid)), + Err(e) => { + log::warn!("[socket-mgr] 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), + ) + } + } + } + None => { + log::warn!("[socket-mgr] 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, + ) + } + } + } + None => { + log::debug!( + "[socket-mgr] 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, + ) + } + }; + + // Emit webhook:response back to the backend + emit_via_channel( + emit_tx, + "webhook:response", + json!({ + "correlationId": response.correlation_id, + "statusCode": response.status_code, + "headers": response.headers, + "body": response.body, + }), + ); + + // Log activity for debugging (frontend polls activity via core RPC) + log::info!( + "[socket-mgr] webhook activity: {} {} → status={}, skill={:?}, tunnel={}", + method, + path, + response.status_code, + resolved_skill_id, + tunnel_name, + ); + + log::debug!( + "[socket-mgr] webhook:response emitted (status={})", + response.status_code, + ); +} + +/// Base64-encode a string (for webhook response bodies). +/// Uses the `STANDARD` alphabet (A-Z, a-z, 0-9, +, /) with `=` padding. +fn base64_encode(input: &str) -> String { + use base64::Engine; + base64::engine::general_purpose::STANDARD.encode(input.as_bytes()) +} + // --------------------------------------------------------------------------- // Utility functions // --------------------------------------------------------------------------- diff --git a/src/openhuman/skills/types.rs b/src/openhuman/skills/types.rs index 9c78568ff..67ac5f214 100644 --- a/src/openhuman/skills/types.rs +++ b/src/openhuman/skills/types.rs @@ -3,6 +3,8 @@ use serde::{Deserialize, Serialize}; use std::collections::HashMap; +use crate::openhuman::webhooks::WebhookResponseData; + /// Status of a running skill instance. #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "lowercase")] @@ -99,6 +101,19 @@ pub enum SkillMessage { /// Load params from frontend (e.g. wallet address for wallet skill). /// Delivered after skill/load RPC; skill may export onLoad(params) to receive them. LoadParams { params: serde_json::Value }, + /// Deliver an incoming webhook request to the skill (targeted, not broadcast). + /// The skill must respond via the oneshot reply with status/headers/body. + WebhookRequest { + correlation_id: String, + method: String, + path: String, + headers: HashMap, + query: HashMap, + body: String, + tunnel_id: String, + tunnel_name: String, + reply: tokio::sync::oneshot::Sender>, + }, } /// Origin of a tool-call request entering the skill runtime. diff --git a/src/openhuman/tunnel/cloudflare.rs b/src/openhuman/tunnel/cloudflare.rs deleted file mode 100644 index d92cbb7cd..000000000 --- a/src/openhuman/tunnel/cloudflare.rs +++ /dev/null @@ -1,141 +0,0 @@ -use super::{kill_shared, new_shared_process, SharedProcess, Tunnel, TunnelProcess}; -use anyhow::{bail, Result}; -use tokio::io::AsyncBufReadExt; -use tokio::process::Command; - -/// Cloudflare Tunnel — wraps the `cloudflared` binary. -/// -/// Requires `cloudflared` installed and a tunnel token from the -/// Cloudflare Zero Trust dashboard. -pub struct CloudflareTunnel { - token: String, - proc: SharedProcess, -} - -impl CloudflareTunnel { - pub fn new(token: String) -> Self { - Self { - token, - proc: new_shared_process(), - } - } -} - -#[async_trait::async_trait] -impl Tunnel for CloudflareTunnel { - fn name(&self) -> &str { - "cloudflare" - } - - async fn start(&self, _local_host: &str, local_port: u16) -> Result { - // cloudflared tunnel --no-autoupdate run --token --url http://localhost: - let mut child = Command::new("cloudflared") - .args([ - "tunnel", - "--no-autoupdate", - "run", - "--token", - &self.token, - "--url", - &format!("http://localhost:{local_port}"), - ]) - .stdout(std::process::Stdio::piped()) - .stderr(std::process::Stdio::piped()) - .kill_on_drop(true) - .spawn()?; - - // Read stderr to find the public URL (cloudflared prints it there) - let stderr = child - .stderr - .take() - .ok_or_else(|| anyhow::anyhow!("Failed to capture cloudflared stderr"))?; - - let mut reader = tokio::io::BufReader::new(stderr).lines(); - let mut public_url = String::new(); - - // Wait up to 30s for the tunnel URL to appear - let deadline = tokio::time::Instant::now() + tokio::time::Duration::from_secs(30); - while tokio::time::Instant::now() < deadline { - let line = - tokio::time::timeout(tokio::time::Duration::from_secs(5), reader.next_line()).await; - - match line { - Ok(Ok(Some(l))) => { - tracing::debug!("cloudflared: {l}"); - // Look for the URL pattern in cloudflared output - if let Some(idx) = l.find("https://") { - let url_part = &l[idx..]; - let end = url_part - .find(|c: char| c.is_whitespace()) - .unwrap_or(url_part.len()); - public_url = url_part[..end].to_string(); - break; - } - } - Ok(Ok(None)) => break, - Ok(Err(e)) => bail!("Error reading cloudflared output: {e}"), - Err(_) => {} // timeout on this line, keep trying - } - } - - if public_url.is_empty() { - child.kill().await.ok(); - bail!("cloudflared did not produce a public URL within 30s. Is the token valid?"); - } - - let mut guard = self.proc.lock().await; - *guard = Some(TunnelProcess { - child, - public_url: public_url.clone(), - }); - - Ok(public_url) - } - - async fn stop(&self) -> Result<()> { - kill_shared(&self.proc).await - } - - async fn health_check(&self) -> bool { - let guard = self.proc.lock().await; - guard.as_ref().is_some_and(|tp| tp.child.id().is_some()) - } - - fn public_url(&self) -> Option { - // Can't block on async lock in a sync fn, so we try_lock - self.proc - .try_lock() - .ok() - .and_then(|g| g.as_ref().map(|tp| tp.public_url.clone())) - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn constructor_stores_token() { - let tunnel = CloudflareTunnel::new("cf-token".into()); - assert_eq!(tunnel.token, "cf-token"); - } - - #[test] - fn public_url_is_none_before_start() { - let tunnel = CloudflareTunnel::new("cf-token".into()); - assert!(tunnel.public_url().is_none()); - } - - #[tokio::test] - async fn stop_without_started_process_is_ok() { - let tunnel = CloudflareTunnel::new("cf-token".into()); - let result = tunnel.stop().await; - assert!(result.is_ok()); - } - - #[tokio::test] - async fn health_check_is_false_before_start() { - let tunnel = CloudflareTunnel::new("cf-token".into()); - assert!(!tunnel.health_check().await); - } -} diff --git a/src/openhuman/tunnel/custom.rs b/src/openhuman/tunnel/custom.rs deleted file mode 100644 index dc74148fc..000000000 --- a/src/openhuman/tunnel/custom.rs +++ /dev/null @@ -1,220 +0,0 @@ -use super::{kill_shared, new_shared_process, SharedProcess, Tunnel, TunnelProcess}; -use anyhow::{bail, Result}; -use tokio::io::AsyncBufReadExt; -use tokio::process::Command; - -/// Custom Tunnel — bring your own tunnel binary. -/// -/// Provide a `start_command` with `{port}` and `{host}` placeholders. -/// Optionally provide a `url_pattern` regex to extract the public URL -/// from stdout, and a `health_url` to poll for liveness. -/// -/// Examples: -/// - `bore local {port} --to bore.pub` -/// - `frp -c /etc/frp/frpc.ini` -/// - `ssh -R 80:localhost:{port} serveo.net` -pub struct CustomTunnel { - start_command: String, - health_url: Option, - url_pattern: Option, - proc: SharedProcess, -} - -impl CustomTunnel { - pub fn new( - start_command: String, - health_url: Option, - url_pattern: Option, - ) -> Self { - Self { - start_command, - health_url, - url_pattern, - proc: new_shared_process(), - } - } -} - -#[async_trait::async_trait] -impl Tunnel for CustomTunnel { - fn name(&self) -> &str { - "custom" - } - - async fn start(&self, local_host: &str, local_port: u16) -> Result { - let cmd = self - .start_command - .replace("{port}", &local_port.to_string()) - .replace("{host}", local_host); - - let parts: Vec<&str> = cmd.split_whitespace().collect(); - if parts.is_empty() { - bail!("Custom tunnel start_command is empty"); - } - - let mut child = Command::new(parts[0]) - .args(&parts[1..]) - .stdout(std::process::Stdio::piped()) - .stderr(std::process::Stdio::piped()) - .kill_on_drop(true) - .spawn()?; - - let mut public_url = format!("http://{local_host}:{local_port}"); - - // If a URL pattern is provided, try to extract the public URL from stdout - if let Some(ref pattern) = self.url_pattern { - if let Some(stdout) = child.stdout.take() { - let mut reader = tokio::io::BufReader::new(stdout).lines(); - let deadline = tokio::time::Instant::now() + tokio::time::Duration::from_secs(15); - - while tokio::time::Instant::now() < deadline { - let line = tokio::time::timeout( - tokio::time::Duration::from_secs(3), - reader.next_line(), - ) - .await; - - match line { - Ok(Ok(Some(l))) => { - tracing::debug!("custom-tunnel: {l}"); - // Simple substring match on the pattern - if l.contains(pattern) - || l.contains("https://") - || l.contains("http://") - { - // Extract URL from the line - if let Some(idx) = l.find("https://") { - let url_part = &l[idx..]; - let end = url_part - .find(|c: char| c.is_whitespace()) - .unwrap_or(url_part.len()); - public_url = url_part[..end].to_string(); - break; - } else if let Some(idx) = l.find("http://") { - let url_part = &l[idx..]; - let end = url_part - .find(|c: char| c.is_whitespace()) - .unwrap_or(url_part.len()); - public_url = url_part[..end].to_string(); - break; - } - } - } - Ok(Ok(None) | Err(_)) => break, - Err(_) => {} - } - } - } - } - - let mut guard = self.proc.lock().await; - *guard = Some(TunnelProcess { - child, - public_url: public_url.clone(), - }); - - Ok(public_url) - } - - async fn stop(&self) -> Result<()> { - kill_shared(&self.proc).await - } - - async fn health_check(&self) -> bool { - // If a health URL is configured, try to reach it - if let Some(ref url) = self.health_url { - return crate::openhuman::config::build_runtime_proxy_client("tunnel.custom") - .get(url) - .timeout(std::time::Duration::from_secs(5)) - .send() - .await - .is_ok(); - } - - // Otherwise check if the process is still alive - let guard = self.proc.lock().await; - guard.as_ref().is_some_and(|tp| tp.child.id().is_some()) - } - - fn public_url(&self) -> Option { - self.proc - .try_lock() - .ok() - .and_then(|g| g.as_ref().map(|tp| tp.public_url.clone())) - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[tokio::test] - async fn start_with_empty_command_returns_error() { - let tunnel = CustomTunnel::new(" ".into(), None, None); - let result = tunnel.start("127.0.0.1", 8080).await; - - assert!(result.is_err()); - assert!(result - .unwrap_err() - .to_string() - .contains("start_command is empty")); - } - - #[tokio::test] - async fn start_without_pattern_returns_local_url() { - let tunnel = CustomTunnel::new("sleep 1".into(), None, None); - - let url = tunnel.start("127.0.0.1", 4455).await.unwrap(); - assert_eq!(url, "http://127.0.0.1:4455"); - assert_eq!( - tunnel.public_url().as_deref(), - Some("http://127.0.0.1:4455") - ); - - tunnel.stop().await.unwrap(); - } - - #[tokio::test] - async fn start_with_pattern_extracts_url() { - let tunnel = CustomTunnel::new( - "echo https://public.example".into(), - None, - Some("public.example".into()), - ); - - let url = tunnel.start("localhost", 9999).await.unwrap(); - - assert_eq!(url, "https://public.example"); - assert_eq!( - tunnel.public_url().as_deref(), - Some("https://public.example") - ); - - tunnel.stop().await.unwrap(); - } - - #[tokio::test] - async fn start_replaces_host_and_port_placeholders() { - let tunnel = CustomTunnel::new( - "echo http://{host}:{port}".into(), - None, - Some("http://".into()), - ); - - let url = tunnel.start("10.1.2.3", 4321).await.unwrap(); - - assert_eq!(url, "http://10.1.2.3:4321"); - tunnel.stop().await.unwrap(); - } - - #[tokio::test] - async fn health_check_with_unreachable_health_url_returns_false() { - let tunnel = CustomTunnel::new( - "sleep 1".into(), - Some("http://127.0.0.1:9/healthz".into()), - None, - ); - - assert!(!tunnel.health_check().await); - } -} diff --git a/src/openhuman/tunnel/mod.rs b/src/openhuman/tunnel/mod.rs deleted file mode 100644 index 1484346ba..000000000 --- a/src/openhuman/tunnel/mod.rs +++ /dev/null @@ -1,14 +0,0 @@ -mod cloudflare; -mod custom; -mod ngrok; -mod none; -pub mod ops; -mod tailscale; - -pub use cloudflare::CloudflareTunnel; -pub use custom::CustomTunnel; -pub use ngrok::NgrokTunnel; -#[allow(unused_imports)] -pub use none::NoneTunnel; -pub use ops::*; -pub use tailscale::TailscaleTunnel; diff --git a/src/openhuman/tunnel/ngrok.rs b/src/openhuman/tunnel/ngrok.rs deleted file mode 100644 index 7d16a11f7..000000000 --- a/src/openhuman/tunnel/ngrok.rs +++ /dev/null @@ -1,151 +0,0 @@ -use super::{kill_shared, new_shared_process, SharedProcess, Tunnel, TunnelProcess}; -use anyhow::{bail, Result}; -use tokio::io::AsyncBufReadExt; -use tokio::process::Command; - -/// ngrok Tunnel — wraps the `ngrok` binary. -/// -/// Requires `ngrok` installed. Optionally set a custom domain -/// (requires ngrok paid plan). -pub struct NgrokTunnel { - auth_token: String, - domain: Option, - proc: SharedProcess, -} - -impl NgrokTunnel { - pub fn new(auth_token: String, domain: Option) -> Self { - Self { - auth_token, - domain, - proc: new_shared_process(), - } - } -} - -#[async_trait::async_trait] -impl Tunnel for NgrokTunnel { - fn name(&self) -> &str { - "ngrok" - } - - async fn start(&self, _local_host: &str, local_port: u16) -> Result { - // Set auth token - Command::new("ngrok") - .args(["config", "add-authtoken", &self.auth_token]) - .output() - .await?; - - // Build command: ngrok http [--domain ] - let mut args = vec!["http".to_string(), local_port.to_string()]; - if let Some(ref domain) = self.domain { - args.push("--domain".into()); - args.push(domain.clone()); - } - // Output log to stdout for URL extraction - args.push("--log".into()); - args.push("stdout".into()); - args.push("--log-format".into()); - args.push("logfmt".into()); - - let mut child = Command::new("ngrok") - .args(&args) - .stdout(std::process::Stdio::piped()) - .stderr(std::process::Stdio::piped()) - .kill_on_drop(true) - .spawn()?; - - let stdout = child - .stdout - .take() - .ok_or_else(|| anyhow::anyhow!("Failed to capture ngrok stdout"))?; - - let mut reader = tokio::io::BufReader::new(stdout).lines(); - let mut public_url = String::new(); - - // Wait up to 15s for the tunnel URL - let deadline = tokio::time::Instant::now() + tokio::time::Duration::from_secs(15); - while tokio::time::Instant::now() < deadline { - let line = - tokio::time::timeout(tokio::time::Duration::from_secs(3), reader.next_line()).await; - - match line { - Ok(Ok(Some(l))) => { - tracing::debug!("ngrok: {l}"); - // ngrok logfmt: url=https://xxxx.ngrok-free.app - if let Some(idx) = l.find("url=https://") { - let url_start = idx + 4; // skip "url=" - let url_part = &l[url_start..]; - let end = url_part - .find(|c: char| c.is_whitespace()) - .unwrap_or(url_part.len()); - public_url = url_part[..end].to_string(); - break; - } - } - Ok(Ok(None)) => break, - Ok(Err(e)) => bail!("Error reading ngrok output: {e}"), - Err(_) => {} - } - } - - if public_url.is_empty() { - child.kill().await.ok(); - bail!("ngrok did not produce a public URL within 15s. Is the auth token valid?"); - } - - let mut guard = self.proc.lock().await; - *guard = Some(TunnelProcess { - child, - public_url: public_url.clone(), - }); - - Ok(public_url) - } - - async fn stop(&self) -> Result<()> { - kill_shared(&self.proc).await - } - - async fn health_check(&self) -> bool { - let guard = self.proc.lock().await; - guard.as_ref().is_some_and(|tp| tp.child.id().is_some()) - } - - fn public_url(&self) -> Option { - self.proc - .try_lock() - .ok() - .and_then(|g| g.as_ref().map(|tp| tp.public_url.clone())) - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn constructor_stores_domain() { - let tunnel = NgrokTunnel::new("ngrok-token".into(), Some("my.ngrok.app".into())); - assert_eq!(tunnel.domain.as_deref(), Some("my.ngrok.app")); - } - - #[test] - fn public_url_is_none_before_start() { - let tunnel = NgrokTunnel::new("ngrok-token".into(), None); - assert!(tunnel.public_url().is_none()); - } - - #[tokio::test] - async fn stop_without_started_process_is_ok() { - let tunnel = NgrokTunnel::new("ngrok-token".into(), None); - let result = tunnel.stop().await; - assert!(result.is_ok()); - } - - #[tokio::test] - async fn health_check_is_false_before_start() { - let tunnel = NgrokTunnel::new("ngrok-token".into(), None); - assert!(!tunnel.health_check().await); - } -} diff --git a/src/openhuman/tunnel/none.rs b/src/openhuman/tunnel/none.rs deleted file mode 100644 index dc7189ab3..000000000 --- a/src/openhuman/tunnel/none.rs +++ /dev/null @@ -1,64 +0,0 @@ -use super::Tunnel; -use anyhow::Result; - -/// No-op tunnel — direct local access, no external exposure. -pub struct NoneTunnel; - -#[async_trait::async_trait] -impl Tunnel for NoneTunnel { - fn name(&self) -> &str { - "none" - } - - async fn start(&self, local_host: &str, local_port: u16) -> Result { - Ok(format!("http://{local_host}:{local_port}")) - } - - async fn stop(&self) -> Result<()> { - Ok(()) - } - - async fn health_check(&self) -> bool { - true - } - - fn public_url(&self) -> Option { - None - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn name_is_none() { - let tunnel = NoneTunnel; - assert_eq!(tunnel.name(), "none"); - } - - #[tokio::test] - async fn start_returns_local_url() { - let tunnel = NoneTunnel; - let url = tunnel.start("127.0.0.1", 7788).await.unwrap(); - assert_eq!(url, "http://127.0.0.1:7788"); - } - - #[tokio::test] - async fn stop_is_noop_success() { - let tunnel = NoneTunnel; - assert!(tunnel.stop().await.is_ok()); - } - - #[tokio::test] - async fn health_check_is_always_true() { - let tunnel = NoneTunnel; - assert!(tunnel.health_check().await); - } - - #[test] - fn public_url_is_always_none() { - let tunnel = NoneTunnel; - assert!(tunnel.public_url().is_none()); - } -} diff --git a/src/openhuman/tunnel/ops.rs b/src/openhuman/tunnel/ops.rs deleted file mode 100644 index 597a52cf6..000000000 --- a/src/openhuman/tunnel/ops.rs +++ /dev/null @@ -1,243 +0,0 @@ -use super::*; - -use crate::openhuman::config::{TailscaleTunnelConfig, TunnelConfig}; -use anyhow::{bail, Result}; -use std::sync::Arc; -use tokio::sync::Mutex; - -// ── Tunnel trait ───────────────────────────────────────────────── - -/// Agnostic tunnel abstraction — bring your own tunnel provider. -/// -/// Implementations wrap an external tunnel binary (cloudflared, tailscale, -/// ngrok, etc.) or a custom command. Callers invoke `start()` with the -/// desired local host/port and `stop()` on shutdown. -#[async_trait::async_trait] -pub trait Tunnel: Send + Sync { - /// Human-readable provider name (e.g. "cloudflare", "tailscale") - fn name(&self) -> &str; - - /// Start the tunnel, exposing `local_host:local_port` externally. - /// Returns the public URL on success. - async fn start(&self, local_host: &str, local_port: u16) -> Result; - - /// Stop the tunnel process gracefully. - async fn stop(&self) -> Result<()>; - - /// Check if the tunnel is still alive. - async fn health_check(&self) -> bool; - - /// Return the public URL if the tunnel is running. - fn public_url(&self) -> Option; -} - -// ── Shared child-process handle ────────────────────────────────── - -/// Wraps a spawned tunnel child process so implementations can share it. -pub(crate) struct TunnelProcess { - pub child: tokio::process::Child, - pub public_url: String, -} - -pub(crate) type SharedProcess = Arc>>; - -pub(crate) fn new_shared_process() -> SharedProcess { - Arc::new(Mutex::new(None)) -} - -/// Kill a shared tunnel process if running. -pub(crate) async fn kill_shared(proc: &SharedProcess) -> Result<()> { - let mut guard = proc.lock().await; - if let Some(ref mut tp) = *guard { - tp.child.kill().await.ok(); - tp.child.wait().await.ok(); - } - *guard = None; - Ok(()) -} - -// ── Factory ────────────────────────────────────────────────────── - -/// Create a tunnel from config. Returns `None` for provider "none". -pub fn create_tunnel(config: &TunnelConfig) -> Result>> { - match config.provider.as_str() { - "none" | "" => Ok(None), - - "cloudflare" => { - let cf = config.cloudflare.as_ref().ok_or_else(|| { - anyhow::anyhow!( - "tunnel.provider = \"cloudflare\" but [tunnel.cloudflare] section is missing" - ) - })?; - Ok(Some(Box::new(CloudflareTunnel::new(cf.token.clone())))) - } - - "tailscale" => { - let ts = config.tailscale.as_ref().unwrap_or(&TailscaleTunnelConfig { - funnel: false, - hostname: None, - }); - Ok(Some(Box::new(TailscaleTunnel::new( - ts.funnel, - ts.hostname.clone(), - )))) - } - - "ngrok" => { - let ng = config.ngrok.as_ref().ok_or_else(|| { - anyhow::anyhow!( - "tunnel.provider = \"ngrok\" but [tunnel.ngrok] section is missing" - ) - })?; - Ok(Some(Box::new(NgrokTunnel::new( - ng.auth_token.clone(), - ng.domain.clone(), - )))) - } - - "custom" => { - let cu = config.custom.as_ref().ok_or_else(|| { - anyhow::anyhow!( - "tunnel.provider = \"custom\" but [tunnel.custom] section is missing" - ) - })?; - Ok(Some(Box::new(CustomTunnel::new( - cu.start_command.clone(), - cu.health_url.clone(), - cu.url_pattern.clone(), - )))) - } - - other => bail!( - "Unknown tunnel provider: \"{other}\". Valid: none, cloudflare, tailscale, ngrok, custom" - ), - } -} - -// ── Tests ──────────────────────────────────────────────────────── - -#[cfg(test)] -mod tests { - use super::*; - use crate::openhuman::config::{CloudflareTunnelConfig, CustomTunnelConfig, NgrokTunnelConfig}; - - /// Helper: assert `create_tunnel` returns an error containing `needle`. - fn assert_tunnel_err(cfg: &TunnelConfig, needle: &str) { - match create_tunnel(cfg) { - Err(e) => assert!( - e.to_string().contains(needle), - "Expected error containing \"{needle}\", got: {e}" - ), - Ok(_) => panic!("Expected error containing \"{needle}\", but got Ok"), - } - } - - #[test] - fn factory_none_returns_none() { - let cfg = TunnelConfig::default(); - let t = create_tunnel(&cfg).unwrap(); - assert!(t.is_none()); - } - - #[test] - fn factory_empty_string_returns_none() { - let cfg = TunnelConfig { - provider: String::new(), - ..TunnelConfig::default() - }; - let t = create_tunnel(&cfg).unwrap(); - assert!(t.is_none()); - } - - #[test] - fn factory_unknown_provider_errors() { - let cfg = TunnelConfig { - provider: "wireguard".into(), - ..TunnelConfig::default() - }; - assert_tunnel_err(&cfg, "Unknown tunnel provider"); - } - - #[test] - fn factory_cloudflare_missing_config_errors() { - let cfg = TunnelConfig { - provider: "cloudflare".into(), - ..TunnelConfig::default() - }; - assert_tunnel_err(&cfg, "[tunnel.cloudflare]"); - } - - #[test] - fn factory_cloudflare_with_config_ok() { - let cfg = TunnelConfig { - provider: "cloudflare".into(), - cloudflare: Some(CloudflareTunnelConfig { - token: "test-token".into(), - }), - ..TunnelConfig::default() - }; - let t = create_tunnel(&cfg).unwrap(); - assert!(t.is_some()); - assert_eq!(t.unwrap().name(), "cloudflare"); - } - - #[test] - fn factory_tailscale_defaults_ok() { - let cfg = TunnelConfig { - provider: "tailscale".into(), - ..TunnelConfig::default() - }; - let t = create_tunnel(&cfg).unwrap(); - assert!(t.is_some()); - assert_eq!(t.unwrap().name(), "tailscale"); - } - - #[test] - fn factory_ngrok_missing_config_errors() { - let cfg = TunnelConfig { - provider: "ngrok".into(), - ..TunnelConfig::default() - }; - assert_tunnel_err(&cfg, "[tunnel.ngrok]"); - } - - #[test] - fn factory_ngrok_with_config_ok() { - let cfg = TunnelConfig { - provider: "ngrok".into(), - ngrok: Some(NgrokTunnelConfig { - auth_token: "tok".into(), - domain: None, - }), - ..TunnelConfig::default() - }; - let t = create_tunnel(&cfg).unwrap(); - assert!(t.is_some()); - assert_eq!(t.unwrap().name(), "ngrok"); - } - - #[test] - fn factory_custom_missing_config_errors() { - let cfg = TunnelConfig { - provider: "custom".into(), - ..TunnelConfig::default() - }; - assert_tunnel_err(&cfg, "[tunnel.custom]"); - } - - #[test] - fn factory_custom_with_config_ok() { - let cfg = TunnelConfig { - provider: "custom".into(), - custom: Some(CustomTunnelConfig { - start_command: "echo tunnel".into(), - health_url: None, - url_pattern: None, - }), - ..TunnelConfig::default() - }; - let t = create_tunnel(&cfg).unwrap(); - assert!(t.is_some()); - assert_eq!(t.unwrap().name(), "custom"); - } -} diff --git a/src/openhuman/tunnel/tailscale.rs b/src/openhuman/tunnel/tailscale.rs deleted file mode 100644 index f983d8e36..000000000 --- a/src/openhuman/tunnel/tailscale.rs +++ /dev/null @@ -1,133 +0,0 @@ -use super::{kill_shared, new_shared_process, SharedProcess, Tunnel, TunnelProcess}; -use anyhow::{bail, Result}; -use tokio::process::Command; - -/// Tailscale Tunnel — uses `tailscale serve` (tailnet-only) or -/// `tailscale funnel` (public internet). -/// -/// Requires Tailscale installed and authenticated (`tailscale up`). -pub struct TailscaleTunnel { - funnel: bool, - hostname: Option, - proc: SharedProcess, -} - -impl TailscaleTunnel { - pub fn new(funnel: bool, hostname: Option) -> Self { - Self { - funnel, - hostname, - proc: new_shared_process(), - } - } -} - -#[async_trait::async_trait] -impl Tunnel for TailscaleTunnel { - fn name(&self) -> &str { - "tailscale" - } - - async fn start(&self, _local_host: &str, local_port: u16) -> Result { - let subcommand = if self.funnel { "funnel" } else { "serve" }; - - // Get the tailscale hostname for URL construction - let hostname = if let Some(ref h) = self.hostname { - h.clone() - } else { - // Query tailscale for the current hostname - let output = Command::new("tailscale") - .args(["status", "--json"]) - .output() - .await?; - - if !output.status.success() { - bail!( - "tailscale status failed: {}", - String::from_utf8_lossy(&output.stderr) - ); - } - - let status: serde_json::Value = - serde_json::from_slice(&output.stdout).unwrap_or_default(); - status["Self"]["DNSName"] - .as_str() - .unwrap_or("localhost") - .trim_end_matches('.') - .to_string() - }; - - // tailscale serve|funnel - let child = Command::new("tailscale") - .args([subcommand, &local_port.to_string()]) - .stdout(std::process::Stdio::piped()) - .stderr(std::process::Stdio::piped()) - .kill_on_drop(true) - .spawn()?; - - let public_url = format!("https://{hostname}:{local_port}"); - - let mut guard = self.proc.lock().await; - *guard = Some(TunnelProcess { - child, - public_url: public_url.clone(), - }); - - Ok(public_url) - } - - async fn stop(&self) -> Result<()> { - // Also reset the tailscale serve/funnel - let subcommand = if self.funnel { "funnel" } else { "serve" }; - Command::new("tailscale") - .args([subcommand, "reset"]) - .output() - .await - .ok(); - - kill_shared(&self.proc).await - } - - async fn health_check(&self) -> bool { - let guard = self.proc.lock().await; - guard.as_ref().is_some_and(|tp| tp.child.id().is_some()) - } - - fn public_url(&self) -> Option { - self.proc - .try_lock() - .ok() - .and_then(|g| g.as_ref().map(|tp| tp.public_url.clone())) - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn constructor_stores_hostname_and_mode() { - let tunnel = TailscaleTunnel::new(true, Some("myhost.tailnet.ts.net".into())); - assert!(tunnel.funnel); - assert_eq!(tunnel.hostname.as_deref(), Some("myhost.tailnet.ts.net")); - } - - #[test] - fn public_url_is_none_before_start() { - let tunnel = TailscaleTunnel::new(false, None); - assert!(tunnel.public_url().is_none()); - } - - #[tokio::test] - async fn health_check_is_false_before_start() { - let tunnel = TailscaleTunnel::new(false, None); - assert!(!tunnel.health_check().await); - } - - #[tokio::test] - async fn stop_without_started_process_is_ok() { - let tunnel = TailscaleTunnel::new(false, None); - let result = tunnel.stop().await; - assert!(result.is_ok()); - } -} diff --git a/src/openhuman/webhooks/mod.rs b/src/openhuman/webhooks/mod.rs new file mode 100644 index 000000000..705d4c95c --- /dev/null +++ b/src/openhuman/webhooks/mod.rs @@ -0,0 +1,11 @@ +//! Webhook tunnel routing — maps backend tunnel UUIDs to owning skills. +//! +//! Routes incoming webhooks from the backend's hosted tunnel system to the +//! appropriate skill. The backend manages tunnel provisioning (ngrok, cloudflare, +//! etc.); this module handles the client-side routing and skill dispatch. + +pub mod router; +pub mod types; + +pub use router::WebhookRouter; +pub use types::{TunnelRegistration, WebhookActivityEntry, WebhookRequest, WebhookResponseData}; diff --git a/src/openhuman/webhooks/router.rs b/src/openhuman/webhooks/router.rs new file mode 100644 index 000000000..f0774af08 --- /dev/null +++ b/src/openhuman/webhooks/router.rs @@ -0,0 +1,310 @@ +//! Webhook router — maps tunnel UUIDs to owning skills with isolation enforcement. + +use super::types::TunnelRegistration; +use log::{debug, error, warn}; +use serde::{Deserialize, Serialize}; +use std::collections::HashMap; +use std::path::PathBuf; +use std::sync::RwLock; + +/// Persistent state serialized to disk. +#[derive(Debug, Default, Serialize, Deserialize)] +struct PersistedRoutes { + registrations: Vec, +} + +/// Routes incoming webhook requests to the skill that owns the tunnel. +/// +/// All mutation methods enforce ownership — a skill can only modify its own +/// tunnel registrations and never see or touch another skill's tunnels. +pub struct WebhookRouter { + /// Keyed by `tunnel_uuid`. + routes: RwLock>, + /// Path to the persistence file (e.g. `~/.openhuman/webhook_routes.json`). + persist_path: Option, +} + +impl WebhookRouter { + /// Create a new router, optionally loading persisted routes from disk. + pub fn new(persist_path: Option) -> Self { + let routes = if let Some(ref path) = persist_path { + match std::fs::read_to_string(path) { + Ok(data) => match serde_json::from_str::(&data) { + Ok(persisted) => { + let map: HashMap = persisted + .registrations + .into_iter() + .map(|r| (r.tunnel_uuid.clone(), r)) + .collect(); + debug!( + "[webhooks] Loaded {} persisted route(s) from {:?}", + map.len(), + path + ); + map + } + Err(e) => { + warn!("[webhooks] Failed to parse persisted routes: {}", e); + HashMap::new() + } + }, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => { + debug!("[webhooks] No persisted routes file at {:?}", path); + HashMap::new() + } + Err(e) => { + error!( + "[webhooks] Failed to read persisted routes at {:?}: {}", + path, e + ); + HashMap::new() + } + } + } else { + HashMap::new() + }; + + Self { + routes: RwLock::new(routes), + persist_path, + } + } + + /// Register a tunnel for a skill. + /// + /// Rejects the operation if the tunnel UUID is already owned by a + /// *different* skill. Re-registering from the same skill is a no-op update. + pub fn register( + &self, + tunnel_uuid: &str, + skill_id: &str, + tunnel_name: Option, + backend_tunnel_id: Option, + ) -> Result<(), String> { + let mut routes = self.routes.write().map_err(|e| e.to_string())?; + + if let Some(existing) = routes.get(tunnel_uuid) { + if existing.skill_id != skill_id { + return Err(format!( + "Tunnel {} is already owned by skill '{}'; skill '{}' cannot register it", + tunnel_uuid, existing.skill_id, skill_id + )); + } + } + + debug!( + "[webhooks] Registering tunnel {} → skill '{}'", + tunnel_uuid, skill_id + ); + + routes.insert( + tunnel_uuid.to_string(), + TunnelRegistration { + tunnel_uuid: tunnel_uuid.to_string(), + skill_id: skill_id.to_string(), + tunnel_name, + backend_tunnel_id, + }, + ); + + drop(routes); + self.persist(); + Ok(()) + } + + /// Unregister a tunnel. Only the owning skill can unregister it. + pub fn unregister(&self, tunnel_uuid: &str, skill_id: &str) -> Result<(), String> { + let mut routes = self.routes.write().map_err(|e| e.to_string())?; + + if let Some(existing) = routes.get(tunnel_uuid) { + if existing.skill_id != skill_id { + return Err(format!( + "Tunnel {} is owned by skill '{}'; skill '{}' cannot unregister it", + tunnel_uuid, existing.skill_id, skill_id + )); + } + debug!( + "[webhooks] Unregistering tunnel {} (skill '{}')", + tunnel_uuid, skill_id + ); + routes.remove(tunnel_uuid); + } else { + debug!( + "[webhooks] Tunnel {} not found for unregister (skill '{}')", + tunnel_uuid, skill_id + ); + } + + drop(routes); + self.persist(); + Ok(()) + } + + /// Remove all tunnel registrations for a skill (called on skill stop/crash). + pub fn unregister_skill(&self, skill_id: &str) { + let mut routes = match self.routes.write() { + Ok(r) => r, + Err(e) => { + warn!("[webhooks] Failed to acquire write lock: {}", e); + return; + } + }; + + let before = routes.len(); + routes.retain(|_, reg| reg.skill_id != skill_id); + let removed = before - routes.len(); + + if removed > 0 { + debug!( + "[webhooks] Unregistered {} tunnel(s) for skill '{}'", + removed, skill_id + ); + drop(routes); + self.persist(); + } + } + + /// Look up which skill owns a tunnel UUID. + pub fn route(&self, tunnel_uuid: &str) -> Option { + self.routes + .read() + .ok()? + .get(tunnel_uuid) + .map(|r| r.skill_id.clone()) + } + + /// List tunnels owned by a specific skill (for the skill JS API). + pub fn list_for_skill(&self, skill_id: &str) -> Vec { + self.routes + .read() + .map(|routes| { + routes + .values() + .filter(|r| r.skill_id == skill_id) + .cloned() + .collect() + }) + .unwrap_or_default() + } + + /// List all tunnel registrations (for the frontend admin UI). + pub fn list_all(&self) -> Vec { + self.routes + .read() + .map(|routes| routes.values().cloned().collect()) + .unwrap_or_default() + } + + /// Persist current routes to disk. + fn persist(&self) { + let Some(ref path) = self.persist_path else { + return; + }; + + // Clone routes under the lock, then release before doing I/O. + let persisted = { + let routes = match self.routes.read() { + Ok(r) => r, + Err(_) => return, + }; + PersistedRoutes { + registrations: routes.values().cloned().collect(), + } + }; + + if let Some(parent) = path.parent() { + let _ = std::fs::create_dir_all(parent); + } + + match serde_json::to_string_pretty(&persisted) { + Ok(json) => { + if let Err(e) = std::fs::write(path, json) { + warn!("[webhooks] Failed to persist routes to {:?}: {}", path, e); + } + } + Err(e) => { + warn!("[webhooks] Failed to serialize routes: {}", e); + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_register_and_route() { + let router = WebhookRouter::new(None); + router + .register("uuid-1", "gmail", Some("Gmail Webhook".into()), None) + .unwrap(); + + assert_eq!(router.route("uuid-1"), Some("gmail".to_string())); + assert_eq!(router.route("uuid-nonexistent"), None); + } + + #[test] + fn test_ownership_enforcement() { + let router = WebhookRouter::new(None); + router + .register("uuid-1", "gmail", Some("Gmail".into()), None) + .unwrap(); + + // Another skill cannot register the same tunnel + let result = router.register("uuid-1", "notion", Some("Notion".into()), None); + assert!(result.is_err()); + assert!(result.unwrap_err().contains("already owned")); + + // Same skill can re-register (update) + router + .register("uuid-1", "gmail", Some("Gmail Updated".into()), None) + .unwrap(); + } + + #[test] + fn test_unregister_ownership() { + let router = WebhookRouter::new(None); + router.register("uuid-1", "gmail", None, None).unwrap(); + + // Another skill cannot unregister + let result = router.unregister("uuid-1", "notion"); + assert!(result.is_err()); + + // Owner can unregister + router.unregister("uuid-1", "gmail").unwrap(); + assert_eq!(router.route("uuid-1"), None); + } + + #[test] + fn test_unregister_skill() { + let router = WebhookRouter::new(None); + router.register("uuid-1", "gmail", None, None).unwrap(); + router.register("uuid-2", "gmail", None, None).unwrap(); + router.register("uuid-3", "notion", None, None).unwrap(); + + router.unregister_skill("gmail"); + + assert_eq!(router.route("uuid-1"), None); + assert_eq!(router.route("uuid-2"), None); + assert_eq!(router.route("uuid-3"), Some("notion".to_string())); + } + + #[test] + fn test_list_for_skill() { + let router = WebhookRouter::new(None); + router.register("uuid-1", "gmail", None, None).unwrap(); + router.register("uuid-2", "notion", None, None).unwrap(); + router.register("uuid-3", "gmail", None, None).unwrap(); + + let gmail_tunnels = router.list_for_skill("gmail"); + assert_eq!(gmail_tunnels.len(), 2); + assert!(gmail_tunnels.iter().all(|t| t.skill_id == "gmail")); + + let notion_tunnels = router.list_for_skill("notion"); + assert_eq!(notion_tunnels.len(), 1); + + let empty = router.list_for_skill("nonexistent"); + assert!(empty.is_empty()); + } +} diff --git a/src/openhuman/webhooks/types.rs b/src/openhuman/webhooks/types.rs new file mode 100644 index 000000000..bdfb2c370 --- /dev/null +++ b/src/openhuman/webhooks/types.rs @@ -0,0 +1,83 @@ +//! Core types for webhook tunnel routing. + +use serde::{Deserialize, Serialize}; +use std::collections::HashMap; + +/// Incoming webhook request forwarded from the backend via Socket.IO. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct WebhookRequest { + /// Correlation ID for request-response matching (e.g. `wh_uuid_ts_hex`). + #[serde(rename = "correlationId")] + pub correlation_id: String, + /// Backend tunnel ID. + #[serde(rename = "tunnelId")] + pub tunnel_id: String, + /// Tunnel UUID (used for routing to the owning skill). + #[serde(rename = "tunnelUuid")] + pub tunnel_uuid: String, + /// Human-readable tunnel name. + #[serde(rename = "tunnelName")] + pub tunnel_name: String, + /// HTTP method (GET, POST, etc.). + pub method: String, + /// Request path after the tunnel prefix. + pub path: String, + /// Request headers. + pub headers: HashMap, + /// Query string parameters. + pub query: HashMap, + /// Base64-encoded request body. + #[serde(default)] + pub body: String, +} + +/// Response data sent back to the backend for a webhook request. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct WebhookResponseData { + /// Must match the incoming request's correlation_id. + #[serde(rename = "correlationId")] + pub correlation_id: String, + /// HTTP status code to return. + #[serde(rename = "statusCode")] + pub status_code: u16, + /// Response headers. + #[serde(default)] + pub headers: HashMap, + /// Base64-encoded response body. + #[serde(default)] + pub body: String, +} + +/// A mapping from a tunnel UUID to the skill that owns it. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct TunnelRegistration { + /// Tunnel UUID (from the backend). + pub tunnel_uuid: String, + /// Skill ID that owns and handles this tunnel. + pub skill_id: String, + /// Human-readable tunnel name (optional, for display). + #[serde(default)] + pub tunnel_name: Option, + /// Backend MongoDB `_id` for CRUD operations. + #[serde(default)] + pub backend_tunnel_id: Option, +} + +/// Entry in the webhook activity log, emitted to the frontend via Tauri events. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct WebhookActivityEntry { + /// Correlation ID of the request. + pub correlation_id: String, + /// Tunnel name. + pub tunnel_name: String, + /// HTTP method. + pub method: String, + /// Request path. + pub path: String, + /// Response status code (None if timed out or no handler). + pub status_code: Option, + /// Skill that handled the request (None if unrouted). + pub skill_id: Option, + /// Unix timestamp in milliseconds. + pub timestamp: u64, +}