stygian_graph/adapters/
graphql.rs

1//! GraphQL API adapter — a generic [`ScrapingService`](crate::ports::ScrapingService) for any spec-compliant
2//! GraphQL endpoint.
3//!
4//! Handles the full request/response lifecycle: query execution, variable
5//! injection, GraphQL error-envelope parsing, Jobber-style cost/throttle
6//! metadata, cursor-based pagination, and pluggable auth strategies.
7//!
8//! Target-specific knowledge (endpoint URL, version headers, default auth) is
9//! supplied by a [`GraphQlTargetPlugin`](crate::ports::graphql_plugin::GraphQlTargetPlugin)
10//! resolved from an optional [`GraphQlPluginRegistry`](crate::application::graphql_plugin_registry::GraphQlPluginRegistry).
11
12use std::collections::HashMap;
13use std::sync::Arc;
14use std::time::Duration;
15
16use async_trait::async_trait;
17use serde_json::{Value, json};
18use tokio::sync::RwLock;
19
20use crate::adapters::graphql_rate_limit::{RequestRateLimit, rate_limit_acquire};
21use crate::adapters::graphql_throttle::{
22    BudgetGuard, PluginBudget, reactive_backoff_ms, update_budget,
23};
24use crate::application::graphql_plugin_registry::GraphQlPluginRegistry;
25use crate::application::pipeline_parser::expand_template;
26use crate::domain::error::{Result, ServiceError, StygianError};
27use crate::ports::auth::ErasedAuthPort;
28use crate::ports::{GraphQlAuth, GraphQlAuthKind, ScrapingService, ServiceInput, ServiceOutput};
29
30// ─────────────────────────────────────────────────────────────────────────────
31// Configuration
32// ─────────────────────────────────────────────────────────────────────────────
33
34/// Configuration for [`GraphQlService`].
35///
36/// # Example
37///
38/// ```rust
39/// use stygian_graph::adapters::graphql::GraphQlConfig;
40///
41/// let config = GraphQlConfig {
42///     timeout_secs: 30,
43///     max_pages: 1000,
44///     user_agent: "stygian-graph/1.0".to_string(),
45/// };
46/// ```
47#[derive(Debug, Clone)]
48pub struct GraphQlConfig {
49    /// Request timeout in seconds (default: 30)
50    pub timeout_secs: u64,
51    /// Maximum number of pages for cursor-paginated queries (default: 1000)
52    pub max_pages: usize,
53    /// User-Agent header sent with every request
54    pub user_agent: String,
55}
56
57impl Default for GraphQlConfig {
58    fn default() -> Self {
59        Self {
60            timeout_secs: 30,
61            max_pages: 1000,
62            user_agent: "stygian-graph/1.0".to_string(),
63        }
64    }
65}
66
67// ─────────────────────────────────────────────────────────────────────────────
68// Adapter
69// ─────────────────────────────────────────────────────────────────────────────
70
71/// `ScrapingService` adapter for GraphQL APIs.
72///
73/// Implement any spec-compliant GraphQL endpoint by constructing a
74/// [`GraphQlService`] with a config and an optional plugin registry. Target
75/// specifics (endpoint, version headers, auth) are supplied either via
76/// `ServiceInput.params` directly or through a registered
77/// [`GraphQlTargetPlugin`](crate::ports::graphql_plugin::GraphQlTargetPlugin).
78///
79/// # Example
80///
81/// ```no_run
82/// use stygian_graph::adapters::graphql::{GraphQlService, GraphQlConfig};
83/// use stygian_graph::ports::{ScrapingService, ServiceInput};
84/// use serde_json::json;
85///
86/// #[tokio::main]
87/// async fn main() -> Result<(), Box<dyn std::error::Error>> {
88///     let service = GraphQlService::new(GraphQlConfig::default(), None);
89///     let input = ServiceInput {
90///         url: "https://countries.trevorblades.com/".to_string(),
91///         params: json!({
92///             "query": "{ countries { code name } }"
93///         }),
94///     };
95///     let output = service.execute(input).await?;
96///     println!("{}", output.data);
97///     Ok(())
98/// }
99/// ```
100pub struct GraphQlService {
101    client: reqwest::Client,
102    config: GraphQlConfig,
103    plugins: Option<Arc<GraphQlPluginRegistry>>,
104    /// Optional runtime auth port — used when no static auth is configured.
105    auth_port: Option<Arc<dyn ErasedAuthPort>>,
106    /// Per-plugin proactive cost-throttle budgets, keyed by plugin name.
107    budgets: Arc<RwLock<HashMap<String, PluginBudget>>>,
108    /// Per-plugin sliding-window request-count rate limiters, keyed by plugin name.
109    rate_limits: Arc<RwLock<HashMap<String, RequestRateLimit>>>,
110}
111
112impl GraphQlService {
113    /// Create a new `GraphQlService`.
114    ///
115    /// `plugins` may be `None` for raw-params mode (no plugin resolution).
116    ///
117    /// # Example
118    ///
119    /// ```no_run
120    /// use stygian_graph::adapters::graphql::{GraphQlService, GraphQlConfig};
121    /// use stygian_graph::ports::ScrapingService;
122    ///
123    /// let service = GraphQlService::new(GraphQlConfig::default(), None);
124    /// assert_eq!(service.name(), "graphql");
125    /// ```
126    pub fn new(config: GraphQlConfig, plugins: Option<Arc<GraphQlPluginRegistry>>) -> Self {
127        let client = reqwest::Client::builder()
128            .timeout(Duration::from_secs(config.timeout_secs))
129            .user_agent(&config.user_agent)
130            .build()
131            .unwrap_or_default();
132        Self {
133            client,
134            config,
135            plugins,
136            auth_port: None,
137            budgets: Arc::new(RwLock::new(HashMap::new())),
138            rate_limits: Arc::new(RwLock::new(HashMap::new())),
139        }
140    }
141
142    /// Attach a runtime auth port.
143    ///
144    /// When set, the port's `erased_resolve_token()` will be called to obtain
145    /// a bearer token whenever `params.auth` is absent and the plugin supplies
146    /// no `default_auth`.
147    ///
148    /// # Example
149    ///
150    /// ```no_run
151    /// use std::sync::Arc;
152    /// use stygian_graph::adapters::graphql::{GraphQlService, GraphQlConfig};
153    /// use stygian_graph::ports::auth::{EnvAuthPort, ErasedAuthPort};
154    ///
155    /// let auth: Arc<dyn ErasedAuthPort> = Arc::new(EnvAuthPort::new("API_TOKEN"));
156    /// let service = GraphQlService::new(GraphQlConfig::default(), None)
157    ///     .with_auth_port(auth);
158    /// ```
159    #[must_use]
160    pub fn with_auth_port(mut self, port: Arc<dyn ErasedAuthPort>) -> Self {
161        self.auth_port = Some(port);
162        self
163    }
164
165    // ── Internal helpers ─────────────────────────────────────────────────────
166
167    /// Apply auth to the request builder.
168    fn apply_auth(builder: reqwest::RequestBuilder, auth: &GraphQlAuth) -> reqwest::RequestBuilder {
169        let token = expand_template(&auth.token);
170        match auth.kind {
171            GraphQlAuthKind::Bearer => builder.header("Authorization", format!("Bearer {token}")),
172            GraphQlAuthKind::ApiKey => builder.header("X-Api-Key", token),
173            GraphQlAuthKind::Header => {
174                let name = auth.header_name.as_deref().unwrap_or("X-Api-Key");
175                builder.header(name, token)
176            }
177            GraphQlAuthKind::None => builder,
178        }
179    }
180
181    /// Parse `GraphQlAuth` from a JSON object like `{"kind":"bearer","token":"..."}`.
182    fn parse_auth(val: &Value) -> Option<GraphQlAuth> {
183        let kind_str = val["kind"].as_str().unwrap_or("none");
184        let kind = match kind_str {
185            "bearer" => GraphQlAuthKind::Bearer,
186            "api_key" => GraphQlAuthKind::ApiKey,
187            "header" => GraphQlAuthKind::Header,
188            _ => GraphQlAuthKind::None,
189        };
190        if kind == GraphQlAuthKind::None {
191            return None;
192        }
193        let token = val["token"].as_str()?.to_string();
194        let header_name = val["header_name"].as_str().map(str::to_string);
195        Some(GraphQlAuth {
196            kind,
197            token,
198            header_name,
199        })
200    }
201
202    /// Check whether the response body indicates throttling.
203    ///
204    /// Returns `Some(retry_after_ms)` on throttle detection via any of:
205    /// 1. `extensions.cost.throttleStatus == "THROTTLED"`
206    /// 2. Any error entry with `extensions.code == "THROTTLED"`
207    /// 3. Any error message containing "throttled" (case-insensitive)
208    #[allow(clippy::indexing_slicing)]
209    fn detect_throttle(body: &Value) -> Option<u64> {
210        // 1. extensions.cost.throttleStatus
211        if body["extensions"]["cost"]["throttleStatus"]
212            .as_str()
213            .is_some_and(|s| s.eq_ignore_ascii_case("THROTTLED"))
214        {
215            return Some(Self::throttle_backoff(body));
216        }
217
218        // 2 & 3. errors array
219        if let Some(errors) = body["errors"].as_array() {
220            for err in errors {
221                if err["extensions"]["code"]
222                    .as_str()
223                    .is_some_and(|c| c.eq_ignore_ascii_case("THROTTLED"))
224                {
225                    return Some(Self::throttle_backoff(body));
226                }
227                if err["message"]
228                    .as_str()
229                    .is_some_and(|m| m.to_ascii_lowercase().contains("throttled"))
230                {
231                    return Some(Self::throttle_backoff(body));
232                }
233            }
234        }
235
236        None
237    }
238
239    /// Calculate retry back-off from `extensions.cost`.
240    ///
241    /// ```text
242    /// deficit = maximumAvailable − currentlyAvailable
243    /// ms      = (deficit / restoreRate * 1000).clamp(500, 2000)
244    /// ```
245    #[allow(
246        clippy::indexing_slicing,
247        clippy::cast_possible_truncation,
248        clippy::cast_sign_loss
249    )]
250    fn throttle_backoff(body: &Value) -> u64 {
251        let cost = &body["extensions"]["cost"];
252        let max_avail = cost["maximumAvailable"].as_f64().unwrap_or(10_000.0);
253        let cur_avail = cost["currentlyAvailable"].as_f64().unwrap_or(0.0);
254        let restore_rate = cost["restoreRate"].as_f64().unwrap_or(500.0);
255        let deficit = (max_avail - cur_avail).max(0.0);
256        let ms = if restore_rate > 0.0 {
257            (deficit / restore_rate * 1000.0) as u64
258        } else {
259            2_000
260        };
261        ms.clamp(500, 2_000)
262    }
263
264    /// Extract the `extensions.cost` object into a metadata-compatible [`Value`].
265    #[allow(clippy::indexing_slicing)]
266    fn extract_cost_metadata(body: &Value) -> Option<Value> {
267        let cost = &body["extensions"]["cost"];
268        if cost.is_null() || cost.is_object() && cost.as_object()?.is_empty() {
269            return None;
270        }
271        Some(cost.clone())
272    }
273
274    /// Navigate a dot-separated JSON path like `"data.clients.pageInfo"`.
275    #[allow(clippy::indexing_slicing)]
276    fn json_path<'v>(root: &'v Value, path: &str) -> &'v Value {
277        let mut cur = root;
278        for key in path.split('.') {
279            cur = &cur[key];
280        }
281        cur
282    }
283
284    /// Execute one GraphQL POST and return the parsed JSON body or an error.
285    #[allow(clippy::indexing_slicing)]
286    async fn post_query(
287        &self,
288        url: &str,
289        query: &str,
290        variables: &Value,
291        operation_name: Option<&str>,
292        auth: Option<&GraphQlAuth>,
293        extra_headers: &HashMap<String, String>,
294    ) -> Result<Value> {
295        let mut body = json!({ "query": query, "variables": variables });
296        if let Some(op) = operation_name {
297            body["operationName"] = json!(op);
298        }
299
300        let mut builder = self
301            .client
302            .post(url)
303            .header("Content-Type", "application/json")
304            .header("Accept", "application/json");
305
306        for (k, v) in extra_headers {
307            builder = builder.header(k.as_str(), v.as_str());
308        }
309
310        if let Some(a) = auth {
311            builder = Self::apply_auth(builder, a);
312        }
313
314        let resp = builder
315            .json(&body)
316            .send()
317            .await
318            .map_err(|e| StygianError::Service(ServiceError::Unavailable(e.to_string())))?;
319
320        let status = resp.status();
321        let text = resp
322            .text()
323            .await
324            .map_err(|e| StygianError::Service(ServiceError::Unavailable(e.to_string())))?;
325
326        if status.as_u16() >= 400 {
327            return Err(StygianError::Service(ServiceError::Unavailable(format!(
328                "HTTP {status}: {text}"
329            ))));
330        }
331
332        serde_json::from_str::<Value>(&text).map_err(|e| {
333            StygianError::Service(ServiceError::InvalidResponse(format!("invalid JSON: {e}")))
334        })
335    }
336
337    /// Validate a parsed GraphQL body (errors array, missing `data` key, throttle).
338    ///
339    /// When a `budget` is supplied, uses `reactive_backoff_ms` (config-aware)
340    /// instead of the fixed-clamp fallback for throttle back-off.  `attempt`
341    /// is the 0-based retry count; callers that implement a retry loop should
342    /// increment it on each attempt to get exponential back-off.
343    #[allow(clippy::indexing_slicing)]
344    fn validate_body(body: &Value, budget: Option<&PluginBudget>, attempt: u32) -> Result<()> {
345        // Throttle check takes priority so callers can retry with backoff.
346        if Self::detect_throttle(body).is_some() {
347            let retry_after_ms = budget.map_or_else(
348                || Self::throttle_backoff(body),
349                |b| reactive_backoff_ms(b.config(), body, attempt),
350            );
351            return Err(StygianError::Service(ServiceError::RateLimited {
352                retry_after_ms,
353            }));
354        }
355
356        if let Some(errors) = body["errors"].as_array()
357            && !errors.is_empty()
358        {
359            let msg = errors[0]["message"]
360                .as_str()
361                .unwrap_or("unknown GraphQL error")
362                .to_string();
363            return Err(StygianError::Service(ServiceError::InvalidResponse(msg)));
364        }
365
366        // `data` key is missing — explicitly null with no errors is allowed (partial response)
367        if body.get("data").is_none() {
368            return Err(StygianError::Service(ServiceError::InvalidResponse(
369                "missing 'data' key in GraphQL response".to_string(),
370            )));
371        }
372
373        Ok(())
374    }
375}
376
377// ─────────────────────────────────────────────────────────────────────────────
378// ScrapingService impl
379// ─────────────────────────────────────────────────────────────────────────────
380
381#[async_trait]
382impl ScrapingService for GraphQlService {
383    fn name(&self) -> &'static str {
384        "graphql"
385    }
386
387    /// Execute a GraphQL query.
388    ///
389    /// Reads `ServiceInput.params` for:
390    /// - `query` (required) — the GraphQL query string
391    /// - `variables` — optional JSON object
392    /// - `operation_name` — optional string
393    /// - `auth` — optional `{"kind": "bearer"|"api_key"|"header"|"none", "token": "..."}`
394    /// - `headers` — optional extra headers object
395    /// - `plugin` — optional plugin name to resolve from the registry
396    /// - `pagination` — optional `{"strategy": "cursor", "page_info_path": "...", "edges_path": "...", "page_size": 50}`
397    ///
398    /// # Errors
399    ///
400    /// Returns `Err` for HTTP ≥ 400, invalid JSON, GraphQL `errors[]`, missing
401    /// `data` key, throttle detection, or pagination runaway.
402    #[allow(clippy::too_many_lines, clippy::indexing_slicing)]
403    async fn execute(&self, input: ServiceInput) -> Result<ServiceOutput> {
404        let params = &input.params;
405
406        // ── 1. Resolve plugin (if any) ────────────────────────────────────
407        let plugin_name = params["plugin"].as_str();
408        let plugin = if let (Some(name), Some(registry)) = (plugin_name, &self.plugins) {
409            Some(registry.get(name)?)
410        } else {
411            None
412        };
413
414        // ── 2. Resolve URL ────────────────────────────────────────────────
415        let url = if !input.url.is_empty() {
416            input.url.clone()
417        } else if let Some(ref p) = plugin {
418            p.endpoint().to_string()
419        } else {
420            return Err(StygianError::Service(ServiceError::Unavailable(
421                "no URL provided and no plugin endpoint available".to_string(),
422            )));
423        };
424
425        // ── 3. Resolve query ──────────────────────────────────────────────
426        let query = params["query"].as_str().ok_or_else(|| {
427            StygianError::Service(ServiceError::InvalidResponse(
428                "params.query is required".to_string(),
429            ))
430        })?;
431
432        let operation_name = params["operation_name"].as_str();
433        let mut variables = params["variables"].clone();
434        if variables.is_null() {
435            variables = json!({});
436        }
437
438        // ── 4. Resolve auth ───────────────────────────────────────────────
439        let auth: Option<GraphQlAuth> = if !params["auth"].is_null() && params["auth"].is_object() {
440            Self::parse_auth(&params["auth"])
441        } else {
442            // Prefer plugin-provided default auth when present; only fall back to
443            // the runtime auth port when the plugin returns None (or no plugin).
444            let plugin_auth = plugin.as_ref().and_then(|p| p.default_auth());
445            if plugin_auth.is_some() {
446                plugin_auth
447            } else if let Some(ref port) = self.auth_port {
448                match port.erased_resolve_token().await {
449                    Ok(token) => Some(GraphQlAuth {
450                        kind: GraphQlAuthKind::Bearer,
451                        token,
452                        header_name: None,
453                    }),
454                    Err(e) => {
455                        let msg = format!("auth port failed to resolve token: {e}");
456                        tracing::error!("{msg}");
457                        return Err(StygianError::Service(ServiceError::AuthenticationFailed(
458                            msg,
459                        )));
460                    }
461                }
462            } else {
463                None
464            }
465        };
466
467        // ── 4b. Lazy-init and acquire per-plugin budget ───────────────────
468        let maybe_budget: Option<PluginBudget> = if let Some(ref p) = plugin {
469            if let Some(throttle_cfg) = p.cost_throttle_config() {
470                let name = p.name().to_string();
471                let budget = {
472                    let read = self.budgets.read().await;
473                    if let Some(b) = read.get(&name) {
474                        b.clone()
475                    } else {
476                        drop(read);
477                        // Slow path: initialise under write lock with double-check
478                        // to prevent two concurrent requests both inserting a fresh
479                        // budget and one overwriting any updates the other has applied.
480                        let mut write = self.budgets.write().await;
481                        write
482                            .entry(name)
483                            .or_insert_with(|| PluginBudget::new(throttle_cfg))
484                            .clone()
485                    }
486                };
487                Some(budget)
488            } else {
489                None
490            }
491        } else {
492            None
493        };
494
495        // ── 4c. Lazy-init per-plugin request-rate limiter ─────────────────
496        let maybe_rl: Option<RequestRateLimit> = if let Some(ref p) = plugin {
497            if let Some(rl_cfg) = p.rate_limit_config() {
498                let name = p.name().to_string();
499                let rl = {
500                    let read = self.rate_limits.read().await;
501                    if let Some(r) = read.get(&name) {
502                        r.clone()
503                    } else {
504                        drop(read);
505                        // Slow path: initialise under write lock with double-check
506                        // to prevent concurrent requests from racing to insert.
507                        let mut write = self.rate_limits.write().await;
508                        write
509                            .entry(name)
510                            .or_insert_with(|| RequestRateLimit::new(rl_cfg))
511                            .clone()
512                    }
513                };
514                Some(rl)
515            } else {
516                None
517            }
518        } else {
519            None
520        };
521
522        // ── 5. Build headers (extra + plugin version headers) ─────────────
523        let mut extra_headers: HashMap<String, String> = params["headers"]
524            .as_object()
525            .map(|obj| {
526                obj.iter()
527                    .filter_map(|(k, v)| v.as_str().map(|s| (k.clone(), s.to_string())))
528                    .collect()
529            })
530            .unwrap_or_default();
531
532        // Plugin version headers override ad-hoc ones for the same key
533        if let Some(ref p) = plugin {
534            for (k, v) in p.version_headers() {
535                extra_headers.insert(k, v);
536            }
537        }
538
539        // ── 6. Resolve pagination config ──────────────────────────────────
540        let pag = &params["pagination"];
541        let use_cursor = pag["strategy"].as_str() == Some("cursor");
542        let page_info_path = pag["page_info_path"]
543            .as_str()
544            .unwrap_or("data.pageInfo")
545            .to_string();
546        let edges_path = pag["edges_path"]
547            .as_str()
548            .unwrap_or("data.edges")
549            .to_string();
550        let page_size: u64 = pag["page_size"]
551            .as_u64()
552            .unwrap_or_else(|| plugin.as_ref().map_or(50, |p| p.default_page_size() as u64));
553
554        // ── 7. Execute (with optional cursor pagination) ───────────────────
555        if use_cursor {
556            // Inject the initial `first`/page-size variable and null cursor
557            variables["first"] = json!(page_size);
558            variables["after"] = json!(null);
559
560            let mut all_edges: Vec<Value> = Vec::new();
561            let mut page = 0usize;
562            let mut cost_meta = json!(null);
563
564            loop {
565                if page >= self.config.max_pages {
566                    return Err(StygianError::Service(ServiceError::InvalidResponse(
567                        format!("pagination exceeded max_pages ({})", self.config.max_pages),
568                    )));
569                }
570
571                // BudgetGuard RAII: reservation is released when `guard`
572                // goes out of scope (via Drop safety-net) or explicitly
573                // via `guard.release().await` on the success path.
574                if let Some(ref rl) = maybe_rl {
575                    rate_limit_acquire(rl).await;
576                }
577                let guard = if let Some(ref b) = maybe_budget {
578                    Some(BudgetGuard::acquire(b).await)
579                } else {
580                    None
581                };
582
583                let body = self
584                    .post_query(
585                        &url,
586                        query,
587                        &variables,
588                        operation_name,
589                        auth.as_ref(),
590                        &extra_headers,
591                    )
592                    .await?;
593
594                Self::validate_body(&body, maybe_budget.as_ref(), 0)?;
595
596                // Update proactive budget from response, then release guard
597                if let Some(ref b) = maybe_budget {
598                    update_budget(b, &body).await;
599                }
600                if let Some(g) = guard {
601                    g.release().await;
602                }
603
604                // Accumulate edges
605                let edges = Self::json_path(&body, &edges_path);
606                if let Some(arr) = edges.as_array() {
607                    all_edges.extend(arr.iter().cloned());
608                }
609
610                // Check for next page
611                let page_info = Self::json_path(&body, &page_info_path);
612                let has_next = page_info["hasNextPage"].as_bool().unwrap_or(false);
613                let end_cursor = page_info["endCursor"].clone();
614
615                cost_meta = Self::extract_cost_metadata(&body).unwrap_or(json!(null));
616                page += 1;
617
618                if !has_next || end_cursor.is_null() {
619                    break;
620                }
621                variables["after"] = end_cursor;
622            }
623
624            let metadata = json!({ "cost": cost_meta, "pages_fetched": page });
625            Ok(ServiceOutput {
626                data: serde_json::to_string(&all_edges).unwrap_or_default(),
627                metadata,
628            })
629        } else {
630            // Single-request mode
631            // BudgetGuard RAII: reservation is released when `guard`
632            // goes out of scope (via Drop safety-net) or explicitly
633            // via `guard.release().await` on the success path.
634            if let Some(ref rl) = maybe_rl {
635                rate_limit_acquire(rl).await;
636            }
637            let guard = if let Some(ref b) = maybe_budget {
638                Some(BudgetGuard::acquire(b).await)
639            } else {
640                None
641            };
642
643            let body = self
644                .post_query(
645                    &url,
646                    query,
647                    &variables,
648                    operation_name,
649                    auth.as_ref(),
650                    &extra_headers,
651                )
652                .await?;
653
654            Self::validate_body(&body, maybe_budget.as_ref(), 0)?;
655
656            // Update proactive budget from response, then release guard
657            if let Some(ref b) = maybe_budget {
658                update_budget(b, &body).await;
659            }
660            if let Some(g) = guard {
661                g.release().await;
662            }
663
664            let cost_meta = Self::extract_cost_metadata(&body).unwrap_or(json!(null));
665            let metadata = json!({ "cost": cost_meta });
666
667            Ok(ServiceOutput {
668                data: serde_json::to_string(&body["data"]).unwrap_or_default(),
669                metadata,
670            })
671        }
672    }
673}
674
675// ─────────────────────────────────────────────────────────────────────────────
676// Tests
677// ─────────────────────────────────────────────────────────────────────────────
678
679#[cfg(test)]
680#[allow(
681    clippy::unwrap_used,
682    clippy::indexing_slicing,
683    clippy::needless_pass_by_value,
684    clippy::field_reassign_with_default,
685    clippy::unnecessary_literal_bound
686)]
687mod tests {
688    use super::*;
689    use std::collections::HashMap;
690    use std::io::Write;
691    use std::sync::Arc;
692
693    use serde_json::json;
694    use tokio::io::{AsyncReadExt, AsyncWriteExt};
695    use tokio::net::TcpListener;
696
697    use crate::application::graphql_plugin_registry::GraphQlPluginRegistry;
698    use crate::ports::graphql_plugin::GraphQlTargetPlugin;
699
700    // ── Mock server ──────────────────────────────────────────────────────────
701
702    /// Minimal HTTP/1.1 mock server that serves one fixed JSON response body.
703    ///
704    /// The server listens on a random port, serves one request, then stops.
705    struct MockGraphQlServer;
706
707    impl MockGraphQlServer {
708        /// Spawn a server that returns HTTP `status` with `body` and run `f`.
709        ///
710        /// The closure receives the base URL `"http://127.0.0.1:<port>"`.
711        async fn run_with<F, Fut>(status: u16, body: impl Into<Vec<u8>>, f: F)
712        where
713            F: FnOnce(String) -> Fut,
714            Fut: std::future::Future<Output = ()>,
715        {
716            let body_bytes: Vec<u8> = body.into();
717            let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
718            let addr = listener.local_addr().unwrap();
719            let url = format!("http://{addr}");
720
721            let body_clone = body_bytes.clone();
722            tokio::spawn(async move {
723                if let Ok((mut stream, _)) = listener.accept().await {
724                    let mut buf = [0u8; 4096];
725                    let _ = stream.read(&mut buf).await;
726                    // Build a minimal HTTP/1.1 response
727                    let mut response = Vec::new();
728                    write!(
729                        response,
730                        "HTTP/1.1 {status} OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
731                        body_clone.len()
732                    ).unwrap();
733                    response.extend_from_slice(&body_clone);
734                    let _ = stream.write_all(&response).await;
735                }
736            });
737
738            f(url).await;
739        }
740
741        /// Variant that captures the received request headers for assertion.
742        async fn run_capturing_request<F, Fut>(body: impl Into<Vec<u8>>, f: F) -> Vec<u8>
743        where
744            F: FnOnce(String) -> Fut,
745            Fut: std::future::Future<Output = ()>,
746        {
747            let body_bytes: Vec<u8> = body.into();
748            let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
749            let addr = listener.local_addr().unwrap();
750            let url = format!("http://{addr}");
751
752            let body_clone = body_bytes.clone();
753            let (tx, mut rx) = tokio::sync::oneshot::channel::<Vec<u8>>();
754            tokio::spawn(async move {
755                if let Ok((mut stream, _)) = listener.accept().await {
756                    let mut buf = vec![0u8; 8192];
757                    let n = stream.read(&mut buf).await.unwrap_or(0);
758                    let request = buf[..n].to_vec();
759                    let _ = tx.send(request);
760
761                    let mut response = Vec::new();
762                    write!(
763                        response,
764                        "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
765                        body_clone.len()
766                    ).unwrap();
767                    response.extend_from_slice(&body_clone);
768                    let _ = stream.write_all(&response).await;
769                }
770            });
771
772            f(url).await;
773
774            rx.try_recv().unwrap_or_default()
775        }
776    }
777
778    fn make_service(plugins: Option<Arc<GraphQlPluginRegistry>>) -> GraphQlService {
779        let mut config = GraphQlConfig::default();
780        config.max_pages = 5; // keep tests fast
781        GraphQlService::new(config, plugins)
782    }
783
784    fn simple_query_body(data: Value) -> Vec<u8> {
785        serde_json::to_vec(&json!({ "data": data })).unwrap()
786    }
787
788    // ── Tests ────────────────────────────────────────────────────────────────
789
790    #[tokio::test]
791    async fn execute_simple_query() {
792        let body = simple_query_body(json!({ "users": [{ "id": 1 }] }));
793        MockGraphQlServer::run_with(200, body, |url| async move {
794            let svc = make_service(None);
795            let input = ServiceInput {
796                url,
797                params: json!({ "query": "{ users { id } }" }),
798            };
799            let output = svc.execute(input).await.unwrap();
800            let data: Value = serde_json::from_str(&output.data).unwrap();
801            assert_eq!(data["users"][0]["id"], 1);
802        })
803        .await;
804    }
805
806    #[tokio::test]
807    async fn graphql_errors_in_200_response() {
808        let body =
809            serde_json::to_vec(&json!({ "errors": [{ "message": "not found" }], "data": null }))
810                .unwrap();
811        MockGraphQlServer::run_with(200, body, |url| async move {
812            let svc = make_service(None);
813            let input = ServiceInput {
814                url,
815                params: json!({ "query": "{ missing }" }),
816            };
817            let err = svc.execute(input).await.unwrap_err();
818            assert!(
819                matches!(err, StygianError::Service(ServiceError::InvalidResponse(_))),
820                "expected InvalidResponse, got {err:?}"
821            );
822        })
823        .await;
824    }
825
826    #[tokio::test]
827    async fn http_error_returns_unavailable() {
828        let body = b"Internal Server Error".to_vec();
829        MockGraphQlServer::run_with(500, body, |url| async move {
830            let svc = make_service(None);
831            let input = ServiceInput {
832                url,
833                params: json!({ "query": "{ x }" }),
834            };
835            let err = svc.execute(input).await.unwrap_err();
836            assert!(
837                matches!(err, StygianError::Service(ServiceError::Unavailable(_))),
838                "expected Unavailable, got {err:?}"
839            );
840        })
841        .await;
842    }
843
844    #[tokio::test]
845    async fn missing_data_key() {
846        let body = serde_json::to_vec(&json!({ "extensions": {} })).unwrap();
847        MockGraphQlServer::run_with(200, body, |url| async move {
848            let svc = make_service(None);
849            let input = ServiceInput {
850                url,
851                params: json!({ "query": "{ x }" }),
852            };
853            let err = svc.execute(input).await.unwrap_err();
854            assert!(
855                matches!(err, StygianError::Service(ServiceError::InvalidResponse(_))),
856                "expected InvalidResponse, got {err:?}"
857            );
858        })
859        .await;
860    }
861
862    #[tokio::test]
863    async fn bearer_auth_header_set() {
864        let body = simple_query_body(json!({}));
865        let request_bytes = MockGraphQlServer::run_capturing_request(body, |url| async move {
866            let svc = make_service(None);
867            let input = ServiceInput {
868                url,
869                params: json!({
870                    "query": "{ x }",
871                    "auth": { "kind": "bearer", "token": "test-token-123" }
872                }),
873            };
874            let _ = svc.execute(input).await;
875        })
876        .await;
877
878        let request_str = String::from_utf8_lossy(&request_bytes);
879        assert!(
880            request_str.contains("authorization: Bearer test-token-123"),
881            "auth header not found in request:\n{request_str}"
882        );
883    }
884
885    #[tokio::test]
886    async fn plugin_version_headers_merged() {
887        struct V1Plugin;
888        impl GraphQlTargetPlugin for V1Plugin {
889            fn name(&self) -> &str {
890                "v1"
891            }
892            fn endpoint(&self) -> &str {
893                "unused"
894            }
895            fn version_headers(&self) -> HashMap<String, String> {
896                [("X-TEST-VERSION".to_string(), "2025-01-01".to_string())].into()
897            }
898        }
899
900        let mut registry = GraphQlPluginRegistry::new();
901        registry.register(Arc::new(V1Plugin));
902
903        let body = simple_query_body(json!({}));
904        let request_bytes = MockGraphQlServer::run_capturing_request(body, |url| async move {
905            let svc = make_service(Some(Arc::new(registry)));
906            let input = ServiceInput {
907                url,
908                params: json!({
909                    "query": "{ x }",
910                    "plugin": "v1"
911                }),
912            };
913            let _ = svc.execute(input).await;
914        })
915        .await;
916
917        let request_str = String::from_utf8_lossy(&request_bytes);
918        assert!(
919            request_str.contains("x-test-version: 2025-01-01"),
920            "version header not found:\n{request_str}"
921        );
922    }
923
924    #[tokio::test]
925    async fn plugin_default_auth_used_when_params_auth_absent() {
926        use crate::ports::{GraphQlAuth, GraphQlAuthKind};
927
928        struct TokenPlugin;
929        impl GraphQlTargetPlugin for TokenPlugin {
930            fn name(&self) -> &str {
931                "tokenplugin"
932            }
933            fn endpoint(&self) -> &str {
934                "unused"
935            }
936            fn default_auth(&self) -> Option<GraphQlAuth> {
937                Some(GraphQlAuth {
938                    kind: GraphQlAuthKind::Bearer,
939                    token: "plugin-default-token".to_string(),
940                    header_name: None,
941                })
942            }
943        }
944
945        let mut registry = GraphQlPluginRegistry::new();
946        registry.register(Arc::new(TokenPlugin));
947
948        let body = simple_query_body(json!({}));
949        let request_bytes = MockGraphQlServer::run_capturing_request(body, |url| async move {
950            let svc = make_service(Some(Arc::new(registry)));
951            let input = ServiceInput {
952                url,
953                // No `auth` field — plugin should supply it
954                params: json!({
955                    "query": "{ x }",
956                    "plugin": "tokenplugin"
957                }),
958            };
959            let _ = svc.execute(input).await;
960        })
961        .await;
962
963        let request_str = String::from_utf8_lossy(&request_bytes);
964        assert!(
965            request_str.contains("Bearer plugin-default-token"),
966            "plugin default auth not applied:\n{request_str}"
967        );
968    }
969
970    #[tokio::test]
971    async fn throttle_response_returns_rate_limited() {
972        let body = serde_json::to_vec(&json!({
973            "data": null,
974            "extensions": {
975                "cost": {
976                    "throttleStatus": "THROTTLED",
977                    "maximumAvailable": 10000,
978                    "currentlyAvailable": 0,
979                    "restoreRate": 500
980                }
981            }
982        }))
983        .unwrap();
984
985        MockGraphQlServer::run_with(200, body, |url| async move {
986            let svc = make_service(None);
987            let input = ServiceInput {
988                url,
989                params: json!({ "query": "{ x }" }),
990            };
991            let err = svc.execute(input).await.unwrap_err();
992            assert!(
993                matches!(
994                    err,
995                    StygianError::Service(ServiceError::RateLimited { retry_after_ms })
996                    if retry_after_ms > 0
997                ),
998                "expected RateLimited, got {err:?}"
999            );
1000        })
1001        .await;
1002    }
1003
1004    #[tokio::test]
1005    async fn cost_metadata_surfaced() {
1006        let body = serde_json::to_vec(&json!({
1007            "data": { "items": [] },
1008            "extensions": {
1009                "cost": {
1010                    "throttleStatus": "PASS",
1011                    "maximumAvailable": 10000,
1012                    "currentlyAvailable": 9800,
1013                    "actualQueryCost": 42,
1014                    "restoreRate": 500
1015                }
1016            }
1017        }))
1018        .unwrap();
1019
1020        MockGraphQlServer::run_with(200, body, |url| async move {
1021            let svc = make_service(None);
1022            let input = ServiceInput {
1023                url,
1024                params: json!({ "query": "{ items { id } }" }),
1025            };
1026            let output = svc.execute(input).await.unwrap();
1027            let cost = &output.metadata["cost"];
1028            assert_eq!(cost["actualQueryCost"], 42);
1029            assert_eq!(cost["throttleStatus"], "PASS");
1030        })
1031        .await;
1032    }
1033
1034    #[tokio::test]
1035    async fn cursor_pagination_accumulates_pages() {
1036        // Two-page scenario: page 1 has next page, page 2 does not.
1037        // We need two independent servers (one per page).
1038        let listener1 = TcpListener::bind("127.0.0.1:0").await.unwrap();
1039        let addr1 = listener1.local_addr().unwrap();
1040        let listener2 = TcpListener::bind("127.0.0.1:0").await.unwrap();
1041        let addr2 = listener2.local_addr().unwrap();
1042
1043        // Both pages go to the same host:port — use a single server that handles
1044        // two sequential connections.
1045        let page1_body = serde_json::to_vec(&json!({
1046            "data": {
1047                "items": {
1048                    "edges": [{"node": {"id": 1}}, {"node": {"id": 2}}],
1049                    "pageInfo": { "hasNextPage": true, "endCursor": "cursor1" }
1050                }
1051            }
1052        }))
1053        .unwrap();
1054
1055        let page2_body = serde_json::to_vec(&json!({
1056            "data": {
1057                "items": {
1058                    "edges": [{"node": {"id": 3}}],
1059                    "pageInfo": { "hasNextPage": false, "endCursor": null }
1060                }
1061            }
1062        }))
1063        .unwrap();
1064
1065        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
1066        let addr = listener.local_addr().unwrap();
1067        let url = format!("http://{addr}");
1068
1069        let bodies = vec![page1_body, page2_body];
1070        tokio::spawn(async move {
1071            for response_body in bodies {
1072                if let Ok((mut stream, _)) = listener.accept().await {
1073                    let mut buf = [0u8; 8192];
1074                    let _ = stream.read(&mut buf).await;
1075                    let mut resp = Vec::new();
1076                    write!(
1077                        resp,
1078                        "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
1079                        response_body.len()
1080                    ).unwrap();
1081                    resp.extend_from_slice(&response_body);
1082                    let _ = stream.write_all(&resp).await;
1083                }
1084            }
1085            // suppress unused warnings — listener1/2 and addr1/2 were created to
1086            // demonstrate the two-listener approach; the actual test uses a single listener
1087            let _ = listener1;
1088            let _ = listener2;
1089            let _ = addr1;
1090            let _ = addr2;
1091        });
1092
1093        let svc = make_service(None);
1094        let input = ServiceInput {
1095            url,
1096            params: json!({
1097                "query": "query($first:Int,$after:String){ items(first:$first,after:$after){ edges{node{id}} pageInfo{hasNextPage endCursor} } }",
1098                "pagination": {
1099                    "strategy": "cursor",
1100                    "page_info_path": "data.items.pageInfo",
1101                    "edges_path": "data.items.edges",
1102                    "page_size": 2
1103                }
1104            }),
1105        };
1106
1107        let output = svc.execute(input).await.unwrap();
1108        let edges: Vec<Value> = serde_json::from_str(&output.data).unwrap();
1109        assert_eq!(edges.len(), 3, "expected 3 accumulated edges");
1110        assert_eq!(edges[0]["node"]["id"], 1);
1111        assert_eq!(edges[2]["node"]["id"], 3);
1112    }
1113
1114    #[tokio::test]
1115    async fn pagination_cap_prevents_infinite_loop() {
1116        // Every page reports hasNextPage=true — the cap should kick in.
1117        let page_body = serde_json::to_vec(&json!({
1118            "data": {
1119                "rows": {
1120                    "edges": [{"node": {"id": 1}}],
1121                    "pageInfo": { "hasNextPage": true, "endCursor": "always-more" }
1122                }
1123            }
1124        }))
1125        .unwrap();
1126
1127        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
1128        let addr = listener.local_addr().unwrap();
1129        let url = format!("http://{addr}");
1130
1131        let page_body_clone = page_body.clone();
1132        tokio::spawn(async move {
1133            while let Ok((mut stream, _)) = listener.accept().await {
1134                let mut buf = [0u8; 8192];
1135                let _ = stream.read(&mut buf).await;
1136                let mut resp = Vec::new();
1137                write!(
1138                    resp,
1139                    "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
1140                    page_body_clone.len()
1141                )
1142                .unwrap();
1143                resp.extend_from_slice(&page_body_clone);
1144                let _ = stream.write_all(&resp).await;
1145            }
1146        });
1147
1148        // max_pages = 5 from make_service
1149        let svc = make_service(None);
1150        let input = ServiceInput {
1151            url,
1152            params: json!({
1153                "query": "{ rows { edges{node{id}} pageInfo{hasNextPage endCursor} } }",
1154                "pagination": {
1155                    "strategy": "cursor",
1156                    "page_info_path": "data.rows.pageInfo",
1157                    "edges_path": "data.rows.edges",
1158                    "page_size": 1
1159                }
1160            }),
1161        };
1162
1163        let err = svc.execute(input).await.unwrap_err();
1164        assert!(
1165            matches!(err, StygianError::Service(ServiceError::InvalidResponse(ref msg)) if msg.contains("max_pages")),
1166            "expected pagination cap error, got {err:?}"
1167        );
1168    }
1169
1170    #[tokio::test]
1171    async fn auth_port_fallback_used_when_no_params_or_plugin_auth() {
1172        use crate::ports::auth::{AuthError, ErasedAuthPort};
1173
1174        struct StaticAuthPort(&'static str);
1175
1176        #[async_trait]
1177        impl ErasedAuthPort for StaticAuthPort {
1178            async fn erased_resolve_token(&self) -> std::result::Result<String, AuthError> {
1179                Ok(self.0.to_string())
1180            }
1181        }
1182
1183        let body = simple_query_body(json!({}));
1184        let request_bytes = MockGraphQlServer::run_capturing_request(body, |url| async move {
1185            let svc = make_service(None).with_auth_port(
1186                Arc::new(StaticAuthPort("port-token-xyz")) as Arc<dyn ErasedAuthPort>
1187            );
1188            let input = ServiceInput {
1189                url,
1190                // No `auth` field and no plugin — auth_port should supply the token
1191                params: json!({ "query": "{ x }" }),
1192            };
1193            let _ = svc.execute(input).await;
1194        })
1195        .await;
1196
1197        let request_str = String::from_utf8_lossy(&request_bytes);
1198        assert!(
1199            request_str.contains("Bearer port-token-xyz"),
1200            "auth_port bearer token not applied:\n{request_str}"
1201        );
1202    }
1203}
stygian_graph/adapters/graphql.rs

stygian_graph/adapters/
graphql.rs
