ryu_webhook_ingress/
tunnels.rs

1//! Concrete [`WebhookIngress`] sources and their enum-dispatch wrapper.
2//!
3//! The project has no `async-trait` dep on this hot path, so the
4//! [`WebhookIngress`] trait declares native `async fn` methods (not object-safe).
5//! Heterogeneous storage is a small closed [`Ingress`] enum, match-dispatched —
6//! never `Box<dyn ..>`. (The kernel-coupling host seam is the one `async-trait`
7//! boundary — see [`super::WebhookIngressHost`].)
8//!
9//! Every source points Composio at Core's **existing** public webhook handler
10//! (`POST /api/composio/webhook`); the tunnel only provides the publicly-reachable
11//! base URL. Core's handler (reached via the host) fires agents unchanged.
12
13use std::process::Stdio;
14use std::sync::RwLock;
15use std::time::Duration;
16
17use anyhow::{anyhow, bail, Result};
18use tokio::io::{AsyncBufReadExt, BufReader};
19use tokio::process::Command;
20
21use super::host::host;
22use super::{IngressKind, WebhookIngress};
23use crate::win_process::NoWindow;
24
25/// The path Composio is pointed at. Every tunnel/relay appends this to its public
26/// base so an inbound webhook lands on Core's existing handler.
27pub const WEBHOOK_PATH: &str = "/api/composio/webhook";
28
29/// Join a public base URL with [`WEBHOOK_PATH`], collapsing a trailing slash so
30/// `https://x.com/` and `https://x.com` both yield `https://x.com/api/composio/webhook`.
31fn join_webhook(base: &str) -> String {
32    format!("{}{}", base.trim_end_matches('/'), WEBHOOK_PATH)
33}
34
35/// **OwnRelay** — the BYO ingress: the user already exposes Core (or a reverse
36/// proxy) at a public URL and configures it here. The base comes from the env
37/// `RYU_WEBHOOK_INGRESS_URL` (preferred) or the value handed at construction
38/// (e.g. a pref). `public_url()` appends the webhook path.
39#[derive(Clone, Debug)]
40pub struct OwnRelaySource {
41    /// The public base URL Core is reachable at (no path). May be empty when
42    /// nothing is configured, in which case `public_url()` errors.
43    pub base_url: String,
44}
45
46/// The env var a BYO operator sets to declare Core's public base URL.
47pub const OWN_RELAY_URL_ENV: &str = "RYU_WEBHOOK_INGRESS_URL";
48
49impl OwnRelaySource {
50    /// Build from the env override first, falling back to the supplied base
51    /// (typically the pref or the resolved local `server_url`).
52    pub fn new(fallback_base: impl Into<String>) -> Self {
53        let env_base = std::env::var(OWN_RELAY_URL_ENV)
54            .ok()
55            .map(|v| v.trim().to_owned())
56            .filter(|v| !v.is_empty());
57        Self {
58            base_url: env_base.unwrap_or_else(|| fallback_base.into()),
59        }
60    }
61}
62
63impl WebhookIngress for OwnRelaySource {
64    fn kind(&self) -> IngressKind {
65        IngressKind::OwnRelay
66    }
67
68    async fn start(&self) -> Result<()> {
69        if self.base_url.trim().is_empty() {
70            bail!(
71                "own-relay ingress: no public URL set (env {OWN_RELAY_URL_ENV} \
72                 or the webhook.ingress.url pref)"
73            );
74        }
75        Ok(())
76    }
77
78    async fn public_url(&self) -> Result<String> {
79        let base = self.base_url.trim();
80        if base.is_empty() {
81            bail!(
82                "own-relay ingress: no public URL set (env {OWN_RELAY_URL_ENV} \
83                 or the webhook.ingress.url pref)"
84            );
85        }
86        Ok(join_webhook(base))
87    }
88}
89
90/// **TailscaleFunnel** — exposes Core's bind port to the public internet via the
91/// P5 mesh's Tailscale Funnel. Consumes the host's `ensure_funnel` / `funnel_url`
92/// (Core forwards to `crate::mesh`). When the mesh is not enabled/available — or
93/// no host is installed — it stub-errors with a clear "Phase 5" message so this
94/// runs standalone.
95#[derive(Clone, Debug)]
96pub struct TailscaleFunnelSource {
97    /// Core's local bind port, the target the Funnel serves.
98    pub port: u16,
99}
100
101impl TailscaleFunnelSource {
102    pub fn new(port: u16) -> Self {
103        Self { port }
104    }
105}
106
107impl WebhookIngress for TailscaleFunnelSource {
108    fn kind(&self) -> IngressKind {
109        IngressKind::TailscaleFunnel
110    }
111
112    async fn start(&self) -> Result<()> {
113        // ensure_funnel itself bails clearly when the mesh is disabled (or no host
114        // is installed); that is the graceful "mesh funnel not available — Phase 5"
115        // path until P5's daemon is enrolled on this node.
116        let url = host()?
117            .ensure_funnel(self.port)
118            .await
119            .map_err(|e| anyhow::anyhow!("mesh funnel not available — Phase 5 ({e})"))?;
120        let _ = url;
121        Ok(())
122    }
123
124    async fn public_url(&self) -> Result<String> {
125        match host()?.funnel_url(self.port).await {
126            Some(base) => Ok(join_webhook(&base)),
127            None => bail!("mesh funnel not available — Phase 5 (no active Funnel for this port)"),
128        }
129    }
130}
131
132/// **Cloudflared** — adopt-or-spawn a `cloudflared` quick tunnel pointed at
133/// Core's local port. `start()` spawns `cloudflared tunnel --url
134/// http://localhost:<port>`, parses the assigned `https://<sub>.trycloudflare.com`
135/// base from its output, and holds the child alive for the process lifetime
136/// (dropping the child tears the tunnel down). `public_url()` returns that base
137/// joined with [`WEBHOOK_PATH`]. No account/login is needed — quick tunnels are
138/// anonymous and ephemeral, which is exactly the BYO-public-URL contract this seam
139/// needs. Requires the `cloudflared` binary on PATH; spawn failure errors clearly.
140#[derive(Clone, Debug)]
141pub struct CloudflaredSource {
142    /// Core's local bind port, the target the tunnel forwards to.
143    pub port: u16,
144}
145
146impl CloudflaredSource {
147    pub fn new(port: u16) -> Self {
148        Self { port }
149    }
150}
151
152/// Process-global state for the single managed cloudflared quick tunnel: the
153/// resolved public base URL plus the held child. The child is kept here (and never
154/// dropped) so the tunnel stays up; `kill_on_drop` ensures it dies with Core.
155struct CloudflaredState {
156    base_url: String,
157    #[allow(dead_code)]
158    child: tokio::process::Child,
159}
160
161static CLOUDFLARED: RwLock<Option<CloudflaredState>> = RwLock::new(None);
162
163/// The current cloudflared base URL, if a tunnel is active.
164fn cloudflared_base_url() -> Option<String> {
165    CLOUDFLARED
166        .read()
167        .ok()
168        .and_then(|g| g.as_ref().map(|s| s.base_url.clone()))
169}
170
171/// Extract a `https://<sub>.trycloudflare.com` URL from a single output line, if
172/// present. cloudflared prints the assigned quick-tunnel URL on its own banner
173/// line (to stderr); this finds it regardless of surrounding box-drawing chars.
174fn extract_trycloudflare_url(line: &str) -> Option<String> {
175    let start = line.find("https://")?;
176    let rest = &line[start..];
177    let end = rest
178        .find(|c: char| c.is_whitespace() || c == '|' || c == '"')
179        .unwrap_or(rest.len());
180    let url = rest[..end].trim_end_matches('/');
181    if url.ends_with(".trycloudflare.com") {
182        Some(url.to_owned())
183    } else {
184        None
185    }
186}
187
188impl WebhookIngress for CloudflaredSource {
189    fn kind(&self) -> IngressKind {
190        IngressKind::Cloudflared
191    }
192
193    async fn start(&self) -> Result<()> {
194        // Idempotent: a tunnel is already up.
195        if cloudflared_base_url().is_some() {
196            return Ok(());
197        }
198
199        let mut child = Command::new("cloudflared")
200            .arg("tunnel")
201            .arg("--no-autoupdate")
202            .arg("--url")
203            .arg(format!("http://localhost:{}", self.port))
204            .stdout(Stdio::piped())
205            .stderr(Stdio::piped())
206            .kill_on_drop(true)
207            .no_window()
208            .spawn()
209            .map_err(|e| {
210                anyhow!(
211                    "cloudflared ingress: failed to spawn `cloudflared` ({e}) — install \
212                     cloudflared and ensure it is on PATH, or use own-relay / tailscale-funnel"
213                )
214            })?;
215
216        // Drain stdout so its pipe never fills (cloudflared logs there too).
217        if let Some(out) = child.stdout.take() {
218            tokio::spawn(async move {
219                let mut lines = BufReader::new(out).lines();
220                while let Ok(Some(_)) = lines.next_line().await {}
221            });
222        }
223
224        // cloudflared prints the assigned URL to stderr. Read until we find it,
225        // hand it back via a oneshot, then keep draining so the pipe never blocks.
226        let stderr = child
227            .stderr
228            .take()
229            .ok_or_else(|| anyhow!("cloudflared ingress: no stderr handle on child"))?;
230        let (tx, rx) = tokio::sync::oneshot::channel::<String>();
231        tokio::spawn(async move {
232            let mut lines = BufReader::new(stderr).lines();
233            let mut tx = Some(tx);
234            while let Ok(Some(line)) = lines.next_line().await {
235                if let Some(url) = extract_trycloudflare_url(&line) {
236                    if let Some(tx) = tx.take() {
237                        let _ = tx.send(url);
238                    }
239                }
240            }
241        });
242
243        let url = tokio::time::timeout(Duration::from_secs(30), rx)
244            .await
245            .map_err(|_| {
246                anyhow!("cloudflared ingress: timed out waiting for the tunnel URL (is cloudflared healthy?)")
247            })?
248            .map_err(|_| {
249                anyhow!("cloudflared ingress: process exited before reporting a tunnel URL")
250            })?;
251
252        if let Ok(mut guard) = CLOUDFLARED.write() {
253            *guard = Some(CloudflaredState {
254                base_url: url,
255                child,
256            });
257        }
258        Ok(())
259    }
260
261    async fn public_url(&self) -> Result<String> {
262        match cloudflared_base_url() {
263            Some(base) => Ok(join_webhook(&base)),
264            None => bail!("cloudflared ingress: no active tunnel (call start first)"),
265        }
266    }
267}
268
269/// **RyuRelay** — the managed push relay (the default). Core opens an outbound
270/// SSE subscription to `apps/server`; Composio POSTs to a public ingress URL and
271/// the server fans the payload out over that stream, which Core dispatches
272/// in-process. The register + SSE-client loop live in [`super::ryu_relay`]; this
273/// source delegates to them.
274#[derive(Clone, Debug, Default)]
275pub struct RyuRelaySource;
276
277impl RyuRelaySource {
278    pub fn new() -> Self {
279        Self
280    }
281}
282
283impl WebhookIngress for RyuRelaySource {
284    fn kind(&self) -> IngressKind {
285        IngressKind::RyuRelay
286    }
287
288    async fn start(&self) -> Result<()> {
289        // Registers with the relay server (publishing the public URL via the
290        // process-global) and spawns the background SSE-client loop. Errors when
291        // not logged in so `main.rs` logs a clear "not active".
292        super::ryu_relay::start().await
293    }
294
295    async fn public_url(&self) -> Result<String> {
296        // The loop publishes the ingress URL to the process-global once register
297        // succeeds; until then there is no URL to report.
298        super::public_url().ok_or_else(|| {
299            anyhow::anyhow!("ryu-relay ingress: not registered yet (login required)")
300        })
301    }
302}
303
304/// The closed set of ingress backends, match-dispatched (no `async-trait`/`dyn`).
305#[derive(Clone, Debug)]
306pub enum Ingress {
307    RyuRelay(RyuRelaySource),
308    TailscaleFunnel(TailscaleFunnelSource),
309    Cloudflared(CloudflaredSource),
310    OwnRelay(OwnRelaySource),
311}
312
313impl Ingress {
314    /// The backend kind this ingress represents.
315    pub fn kind(&self) -> IngressKind {
316        match self {
317            Ingress::RyuRelay(s) => s.kind(),
318            Ingress::TailscaleFunnel(s) => s.kind(),
319            Ingress::Cloudflared(s) => s.kind(),
320            Ingress::OwnRelay(s) => s.kind(),
321        }
322    }
323
324    /// Start (or adopt) the backend so it is ready to receive webhooks.
325    pub async fn start(&self) -> Result<()> {
326        match self {
327            Ingress::RyuRelay(s) => s.start().await,
328            Ingress::TailscaleFunnel(s) => s.start().await,
329            Ingress::Cloudflared(s) => s.start().await,
330            Ingress::OwnRelay(s) => s.start().await,
331        }
332    }
333
334    /// The public URL Composio should be pointed at (ends in [`WEBHOOK_PATH`]).
335    pub async fn public_url(&self) -> Result<String> {
336        match self {
337            Ingress::RyuRelay(s) => s.public_url().await,
338            Ingress::TailscaleFunnel(s) => s.public_url().await,
339            Ingress::Cloudflared(s) => s.public_url().await,
340            Ingress::OwnRelay(s) => s.public_url().await,
341        }
342    }
343}
344
345#[cfg(test)]
346mod tests {
347    use super::*;
348
349    #[test]
350    fn join_webhook_strips_trailing_slash() {
351        assert_eq!(
352            join_webhook("https://x.com"),
353            "https://x.com/api/composio/webhook"
354        );
355        assert_eq!(
356            join_webhook("https://x.com/"),
357            "https://x.com/api/composio/webhook"
358        );
359    }
360
361    #[tokio::test]
362    async fn own_relay_public_url_appends_path() {
363        let src = OwnRelaySource {
364            base_url: "https://relay.example.com/".to_owned(),
365        };
366        assert_eq!(
367            src.public_url().await.unwrap(),
368            "https://relay.example.com/api/composio/webhook"
369        );
370        assert_eq!(src.kind(), IngressKind::OwnRelay);
371    }
372
373    #[tokio::test]
374    async fn own_relay_empty_base_errors() {
375        let src = OwnRelaySource {
376            base_url: "   ".to_owned(),
377        };
378        assert!(src.public_url().await.is_err());
379        assert!(src.start().await.is_err());
380    }
381
382    #[tokio::test]
383    async fn ryu_relay_kind_is_ryu_relay() {
384        // Network-free: do NOT call start() (it would register + spawn the SSE
385        // loop against the live relay server when a ~/.ryu/auth.json token exists,
386        // which is the case on a developer machine). public_url() reads the
387        // process-global PUBLIC_URL, which other tests mutate in parallel, so it
388        // is not asserted here. The frame-parser + register logic is unit-tested
389        // in super::ryu_relay.
390        let src = RyuRelaySource::new();
391        assert_eq!(src.kind(), IngressKind::RyuRelay);
392    }
393
394    #[test]
395    fn extract_trycloudflare_url_parses_banner() {
396        // The real banner wraps the URL in box-drawing chars; parsing must ignore them.
397        assert_eq!(
398            extract_trycloudflare_url(
399                "2024-01-01 INF |  https://random-words-1234.trycloudflare.com  |"
400            ),
401            Some("https://random-words-1234.trycloudflare.com".to_owned())
402        );
403        // Trailing slash is collapsed.
404        assert_eq!(
405            extract_trycloudflare_url("https://abc.trycloudflare.com/"),
406            Some("https://abc.trycloudflare.com".to_owned())
407        );
408        // A non-trycloudflare https URL (e.g. the docs link cloudflared prints) is ignored.
409        assert_eq!(
410            extract_trycloudflare_url("Visit https://developers.cloudflare.com for docs"),
411            None
412        );
413        // Lines without a URL yield nothing.
414        assert_eq!(extract_trycloudflare_url("starting tunnel"), None);
415    }
416
417    #[tokio::test]
418    async fn cloudflared_public_url_errors_without_tunnel() {
419        // public_url() is deterministic + network-free: with no active tunnel it
420        // errors. start() is NOT called here — on a dev machine that has
421        // cloudflared on PATH it would spawn a real anonymous tunnel, which a unit
422        // test must never do. The spawn-failure path is covered below only when the
423        // binary is absent.
424        let src = CloudflaredSource::new(7980);
425        assert_eq!(src.kind(), IngressKind::Cloudflared);
426        if cloudflared_base_url().is_none() {
427            assert!(src.public_url().await.is_err());
428        }
429    }
430
431    #[tokio::test]
432    async fn cloudflared_start_errors_when_binary_absent() {
433        // Only exercise start() when `cloudflared` is NOT installed, so the test
434        // asserts the graceful spawn-failure error without ever opening a live
435        // tunnel on a machine that happens to have the binary.
436        let has_binary = std::process::Command::new("cloudflared")
437            .arg("--version")
438            .stdout(std::process::Stdio::null())
439            .stderr(std::process::Stdio::null())
440            .no_window()
441            .status()
442            .is_ok();
443        if !has_binary {
444            let src = CloudflaredSource::new(7980);
445            assert!(src.start().await.is_err());
446        }
447    }
448
449    #[tokio::test]
450    async fn tailscale_funnel_stub_errors_when_mesh_off() {
451        // In the test process RYU_MESH_ENABLED is unset → mesh off → both paths
452        // surface the clear Phase-5 stub error rather than panicking.
453        if std::env::var("RYU_MESH_ENABLED").is_err() {
454            let src = TailscaleFunnelSource::new(7980);
455            assert_eq!(src.kind(), IngressKind::TailscaleFunnel);
456            assert!(src.start().await.is_err());
457            assert!(src.public_url().await.is_err());
458        }
459    }
460
461    #[tokio::test]
462    async fn enum_dispatch_routes_to_variant() {
463        let ing = Ingress::OwnRelay(OwnRelaySource {
464            base_url: "https://x.com".to_owned(),
465        });
466        assert_eq!(ing.kind(), IngressKind::OwnRelay);
467        assert_eq!(
468            ing.public_url().await.unwrap(),
469            "https://x.com/api/composio/webhook"
470        );
471        assert!(ing.start().await.is_ok());
472
473        let relay = Ingress::RyuRelay(RyuRelaySource::new());
474        assert_eq!(relay.kind(), IngressKind::RyuRelay);
475    }
476}
ryu_webhook_ingress/tunnels.rs

ryu_webhook_ingress/
tunnels.rs
