gthings_cdp/
session.rs

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

gthings_cdp/
session.rs
