gthings_cdp/
session.rs

1use crate::connection::{CdpEvent, Connection};
2use crate::error::{CdpError, Result};
3use crate::tab::Tab;
4use gthings_common::domain_reputation::QualityFlag;
5use serde_json::Value;
6use std::time::Duration;
7use tokio::sync::broadcast;
8
9/// High-level CDP session. Manages connection + tabs with event-driven lifecycle.
10pub struct Session {
11    conn: Connection,
12}
13
14impl std::fmt::Debug for Session {
15    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
16        f.debug_struct("Session").finish_non_exhaustive()
17    }
18}
19
20/// Wait for a specific CDP event on a pre-subscribed receiver.
21/// Subscribe BEFORE the triggering action to avoid missing the event.
22async fn wait_for_event(
23    rx: &mut broadcast::Receiver<CdpEvent>,
24    method: &str,
25    predicate: impl Fn(&CdpEvent) -> bool + Send,
26    timeout: Duration,
27) -> Result<CdpEvent> {
28    tokio::time::timeout(timeout, async move {
29        loop {
30            match rx.recv().await {
31                Ok(event) if event.method.as_str() == method && predicate(&event) => {
32                    return Ok(event);
33                }
34                Ok(_) => continue,
35                Err(broadcast::error::RecvError::Closed) => {
36                    return Err(CdpError::ConnectionFailed {
37                        detail: "event channel closed while waiting".into(),
38                    });
39                }
40                Err(broadcast::error::RecvError::Lagged(n)) => {
41                    tracing::warn!("Event receiver lagged by {n} messages");
42                    continue;
43                }
44            }
45        }
46    })
47    .await
48    .map_err(|_| CdpError::NavigationTimeout {
49        url: "unknown".into(),
50        timeout: timeout.as_secs(),
51    })?
52}
53
54impl Session {
55    /// Connect to browser via WebSocket URL
56    pub async fn connect(ws_url: &str) -> Result<Self> {
57        let conn = Connection::connect(ws_url).await?;
58        Ok(Session { conn })
59    }
60
61    /// Create a new tab/page
62    pub async fn create_tab(&self, url: &str) -> Result<Tab> {
63        Tab::create(self, url).await
64    }
65
66    /// Evaluate JavaScript in a tab, return JSON result
67    pub async fn evaluate(&self, tab: &Tab, js: &str) -> Result<Value> {
68        tab.evaluate(self, js).await
69    }
70
71    /// Navigate to URL and wait for networkIdle lifecycle event
72    pub async fn navigate(&self, tab: &Tab, url: &str) -> Result<()> {
73        let conn = &self.conn;
74        let sid = tab.session_id.as_deref();
75
76        // 1. Enable Page events so we receive lifecycle events
77        conn.call("Page.enable", serde_json::json!({}), sid).await?;
78
79        // 1a. Enable lifecycle events (required for Chrome 144+)
80        conn.call(
81            "Page.setLifecycleEventsEnabled",
82            serde_json::json!({"enabled": true}),
83            sid,
84        )
85        .await?;
86
87        // 2. Subscribe BEFORE navigation — don't miss the networkIdle event
88        let mut rx = conn.event_rx();
89
90        // 3. Start navigation
91        conn.call("Page.navigate", serde_json::json!({"url": url}), sid)
92            .await?;
93
94        // 4. Wait for networkIdle using the pre-subscribed receiver
95        let result = wait_for_event(
96            &mut rx,
97            "Page.lifecycleEvent",
98            |evt| evt.params.get("name").and_then(|v| v.as_str()) == Some("networkIdle"),
99            Duration::from_secs(10),
100        )
101        .await;
102
103        // Fallback: if lifecycle event timed out, poll document.readyState
104        match result {
105            Ok(_) => {}
106            Err(CdpError::NavigationTimeout { .. }) => {
107                tracing::warn!("Lifecycle event timeout, falling back to readyState polling");
108                // Poll document.readyState up to 5 more seconds
109                for _ in 0..10 {
110                    if let Ok(val) = conn
111                        .call(
112                            "Runtime.evaluate",
113                            serde_json::json!({
114                                "expression": "document.readyState",
115                                "returnByValue": true
116                            }),
117                            sid,
118                        )
119                        .await
120                    {
121                        if val
122                            .get("result")
123                            .and_then(|r| r.get("value"))
124                            .and_then(|v| v.as_str())
125                            == Some("complete")
126                        {
127                            return Ok(());
128                        }
129                    }
130                    tokio::time::sleep(Duration::from_millis(500)).await;
131                }
132                return Err(CdpError::NavigationTimeout {
133                    url: url.to_string(),
134                    timeout: 15,
135                });
136            }
137            Err(e) => return Err(e),
138        }
139
140        Ok(())
141    }
142
143    /// Runs a compact JS snippet in the page to detect quality issues before extraction.
144    /// The JS snippet checks for Cloudflare/Turnstile bot walls, reCAPTCHA/hCaptcha, and paywall text markers.
145    /// Keep JS logic in sync with `Session::parse_signal_flags` and `gthings_extraction::quality::detection`.
146    pub async fn check_page_signals(&self, tab: &Tab) -> Result<Vec<QualityFlag>> {
147        let js = r#"
148            (() => {
149                const flags = [];
150                if (document.querySelector('#cf-challenge, .cf-turnstile, [class*="challenge"], [id*="challenge"]'))
151                    flags.push("BotWall");
152                if (document.title.toLowerCase().includes("just a moment"))
153                    flags.push("BotWall");
154                if (document.querySelector('iframe[src*="recaptcha"], iframe[src*="hcaptcha"], .h-captcha, .g-recaptcha'))
155                    flags.push("Captcha");
156                const text = (document.body?.innerText || '').slice(0, 2000).toLowerCase();
157                if (/subscribe to continue|sign in to read|you have reached your free article limit|subscribe to read|log in to read this/i.test(text))
158                    flags.push("Paywall");
159                return flags;
160            })()
161        "#;
162
163        let result = tab.evaluate(self, js).await?;
164        Ok(Self::parse_signal_flags(&result))
165    }
166
167    /// Wait for a CDP event matching method + predicate.
168    ///
169    /// Warning: Creates a new event subscription. Subscribe BEFORE the action that
170    /// triggers the event to avoid race conditions. For navigation, use
171    /// [`navigate()`](Session::navigate) instead.
172    pub async fn wait_for<F>(
173        &self,
174        method: &str,
175        predicate: F,
176        timeout: Duration,
177    ) -> Result<CdpEvent>
178    where
179        F: Fn(&CdpEvent) -> bool + Send + 'static,
180    {
181        let mut rx = self.conn.event_rx();
182
183        tokio::time::timeout(timeout, async move {
184            loop {
185                match rx.recv().await {
186                    Ok(event) if event.method.as_str() == method && predicate(&event) => {
187                        return Ok(event);
188                    }
189                    Ok(_) => continue,
190                    Err(broadcast::error::RecvError::Closed) => {
191                        return Err(CdpError::ConnectionFailed {
192                            detail: "event channel closed while waiting".into(),
193                        });
194                    }
195                    Err(broadcast::error::RecvError::Lagged(n)) => {
196                        tracing::warn!("Event receiver lagged by {n} messages");
197                        continue;
198                    }
199                }
200            }
201        })
202        .await
203        .map_err(|_| CdpError::CdpCallFailed {
204            method: format!("wait_for({method})"),
205            detail: format!("timeout after {timeout:?}"),
206        })?
207    }
208
209    /// Close a tab
210    pub async fn close_tab(&self, tab: Tab) -> Result<()> {
211        tab.close(self).await
212    }
213
214    /// Disconnect from browser
215    pub async fn disconnect(self) -> Result<()> {
216        self.conn.close().await;
217        Ok(())
218    }
219
220    /// Access the underlying Connection (for direct CDP calls)
221    pub fn connection(&self) -> &Connection {
222        &self.conn
223    }
224
225    /// Parse a `Runtime.evaluate` result value into quality flags.
226    ///
227    /// Public and crate-visible for testing. The JS snippet returns an array
228    /// of strings like `["BotWall", "Captcha"]`.
229    pub(crate) fn parse_signal_flags(value: &Value) -> Vec<QualityFlag> {
230        match value
231            .get("result")
232            .and_then(|r| r.get("value"))
233            .and_then(|v| v.as_array())
234        {
235            Some(arr) => arr
236                .iter()
237                .filter_map(|v| {
238                    v.as_str().and_then(|s| match s {
239                        "BotWall" => Some(QualityFlag::BotWall),
240                        "Captcha" => Some(QualityFlag::Captcha),
241                        "Paywall" => Some(QualityFlag::Paywall),
242                        "EmptyShell" => Some(QualityFlag::EmptyShell),
243                        "Garbled" => Some(QualityFlag::Garbled),
244                        "ThinContent" => Some(QualityFlag::ThinContent),
245                        "Truncated" => Some(QualityFlag::Truncated),
246                        _ => None,
247                    })
248                })
249                .collect(),
250            None => Vec::new(),
251        }
252    }
253}
254
255#[cfg(test)]
256mod tests {
257    use super::*;
258    use serde_json::json;
259
260    #[test]
261    fn test_parse_signal_flags_empty() {
262        let val = json!({"result": {"type": "object", "value": []}});
263        let flags = Session::parse_signal_flags(&val);
264        assert!(flags.is_empty());
265    }
266
267    #[test]
268    fn test_parse_signal_flags_botwall() {
269        let val = json!({"result": {"type": "object", "value": ["BotWall"]}});
270        let flags = Session::parse_signal_flags(&val);
271        assert_eq!(flags, vec![QualityFlag::BotWall]);
272    }
273
274    #[test]
275    fn test_parse_signal_flags_multiple() {
276        let val = json!({"result": {"type": "object", "value": ["BotWall", "Captcha"]}});
277        let flags = Session::parse_signal_flags(&val);
278        assert_eq!(flags, vec![QualityFlag::BotWall, QualityFlag::Captcha]);
279    }
280
281    #[test]
282    fn test_parse_signal_flags_paywall() {
283        let val = json!({"result": {"type": "object", "value": ["Paywall"]}});
284        let flags = Session::parse_signal_flags(&val);
285        assert_eq!(flags, vec![QualityFlag::Paywall]);
286    }
287
288    #[test]
289    fn test_parse_signal_flags_unknown_ignored() {
290        let val =
291            json!({"result": {"type": "object", "value": ["BotWall", "UnknownFlag", "Captcha"]}});
292        let flags = Session::parse_signal_flags(&val);
293        assert_eq!(flags, vec![QualityFlag::BotWall, QualityFlag::Captcha]);
294    }
295
296    #[test]
297    fn test_parse_signal_flags_missing_result() {
298        let val = json!({});
299        let flags = Session::parse_signal_flags(&val);
300        assert!(flags.is_empty());
301    }
302
303    #[test]
304    fn test_parse_signal_flags_non_array_value() {
305        let val = json!({"result": {"type": "string", "value": "not_an_array"}});
306        let flags = Session::parse_signal_flags(&val);
307        assert!(flags.is_empty());
308    }
309}
gthings_cdp/session.rs

gthings_cdp/
session.rs
