From 1b131baf70eaf2e54400872dc1f48381961c16fa Mon Sep 17 00:00:00 2001 From: Steven Enamakel <31011319+senamakel@users.noreply.github.com> Date: Tue, 31 Mar 2026 20:19:04 -0700 Subject: [PATCH] feat(webhooks): webhook tunnel routing for skills + remove legacy tunnel module (#147) * feat(webhooks): implement webhook management interface and routing - Added a new Webhooks page with TunnelList and WebhookActivity components for managing webhook tunnels and displaying recent activity. - Introduced useWebhooks hook for handling CRUD operations related to tunnels, including fetching, creating, and deleting tunnels. - Implemented a WebhookRouter in the backend to route incoming webhook requests to the appropriate skills based on tunnel UUIDs. - Enhanced the API for tunnel management, including the ability to register and unregister tunnels for specific skills. - Updated the Redux store to manage webhooks state, including tunnels, registrations, and activity logs. This update provides a comprehensive interface for managing webhooks, improving the overall functionality and user experience in handling webhook events. * refactor(tunnel): remove tunnel-related modules and configurations - Deleted tunnel-related modules including Cloudflare, Custom, Ngrok, and Tailscale, along with their associated configurations and implementations. - Removed references to TunnelConfig and related functions from the configuration and schema files. - Cleaned up the mod.rs files to reflect the removal of tunnel modules, streamlining the codebase. This refactor simplifies the project structure by eliminating unused tunnel functionalities, enhancing maintainability and clarity. * refactor(config): remove tunnel settings from schemas and controllers - Eliminated the `update_tunnel_settings` controller and its associated schema from the configuration files. - Streamlined the `all_registered_controllers` function by removing the handler for tunnel settings, enhancing code clarity and maintainability. This refactor simplifies the configuration structure by removing unused tunnel-related functionalities. * refactor(tunnel): remove tunnel settings and related configurations - Eliminated tunnel-related state variables and functions from the TauriCommandsPanel component, streamlining the settings interface. - Removed the `openhumanUpdateTunnelSettings` function and `TunnelConfig` interface from the utility commands, enhancing code clarity. - Updated the core RPC client to remove legacy tunnel method aliases, further simplifying the codebase. This refactor focuses on cleaning up unused tunnel functionalities, improving maintainability and clarity across the application. * style: apply prettier and cargo fmt formatting Co-Authored-By: Claude Opus 4.6 (1M context) --------- Co-authored-by: Claude Opus 4.6 (1M context) --- app/src/AppRoutes.tsx | 10 + .../settings/panels/TauriCommandsPanel.tsx | 100 +----- app/src/components/webhooks/TunnelList.tsx | 219 +++++++++++++ .../components/webhooks/WebhookActivity.tsx | 83 +++++ app/src/hooks/useWebhooks.ts | 93 ++++++ app/src/pages/Webhooks.tsx | 48 +++ app/src/services/api/tunnelsApi.ts | 21 +- app/src/services/coreRpcClient.ts | 1 - app/src/store/index.ts | 2 + app/src/store/webhooksSlice.ts | 92 ++++++ app/src/utils/tauriCommands.ts | 24 -- src/openhuman/config/mod.rs | 19 +- src/openhuman/config/ops.rs | 25 +- src/openhuman/config/schema/mod.rs | 6 - src/openhuman/config/schema/tunnel.rs | 54 --- src/openhuman/config/schema/types.rs | 4 - src/openhuman/config/schemas.rs | 46 --- src/openhuman/config/settings_cli.rs | 4 - src/openhuman/mod.rs | 2 +- src/openhuman/skills/qjs_engine.rs | 17 + .../skills/qjs_skill_instance/event_loop.rs | 82 +++++ .../skills/qjs_skill_instance/instance.rs | 1 + .../skills/qjs_skill_instance/types.rs | 1 + .../skills/quickjs_libs/bootstrap.js | 128 ++++++++ .../skills/quickjs_libs/qjs_ops/mod.rs | 1 + .../skills/quickjs_libs/qjs_ops/ops.rs | 1 + .../quickjs_libs/qjs_ops/ops_webhook.rs | 83 +++++ .../skills/quickjs_libs/qjs_ops/types.rs | 1 + src/openhuman/skills/skill_registry.rs | 61 ++++ src/openhuman/skills/socket_manager.rs | 181 ++++++++++ src/openhuman/skills/types.rs | 15 + src/openhuman/tunnel/cloudflare.rs | 141 -------- src/openhuman/tunnel/custom.rs | 220 ------------- src/openhuman/tunnel/mod.rs | 14 - src/openhuman/tunnel/ngrok.rs | 151 --------- src/openhuman/tunnel/none.rs | 64 ---- src/openhuman/tunnel/ops.rs | 243 -------------- src/openhuman/tunnel/tailscale.rs | 133 -------- src/openhuman/webhooks/mod.rs | 11 + src/openhuman/webhooks/router.rs | 310 ++++++++++++++++++ src/openhuman/webhooks/types.rs | 83 +++++ 41 files changed, 1547 insertions(+), 1248 deletions(-) create mode 100644 app/src/components/webhooks/TunnelList.tsx create mode 100644 app/src/components/webhooks/WebhookActivity.tsx create mode 100644 app/src/hooks/useWebhooks.ts create mode 100644 app/src/pages/Webhooks.tsx create mode 100644 app/src/store/webhooksSlice.ts delete mode 100644 src/openhuman/config/schema/tunnel.rs create mode 100644 src/openhuman/skills/quickjs_libs/qjs_ops/ops_webhook.rs delete mode 100644 src/openhuman/tunnel/cloudflare.rs delete mode 100644 src/openhuman/tunnel/custom.rs delete mode 100644 src/openhuman/tunnel/mod.rs delete mode 100644 src/openhuman/tunnel/ngrok.rs delete mode 100644 src/openhuman/tunnel/none.rs delete mode 100644 src/openhuman/tunnel/ops.rs delete mode 100644 src/openhuman/tunnel/tailscale.rs create mode 100644 src/openhuman/webhooks/mod.rs create mode 100644 src/openhuman/webhooks/router.rs create mode 100644 src/openhuman/webhooks/types.rs 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, +}