faucet_source_graphql/
config.rs

1//! GraphQL source configuration.
2
3use faucet_core::{
4    AuthSpec, DEFAULT_BATCH_SIZE, FaucetError, TlsClientConfig, validate_batch_size,
5};
6use reqwest::header::HeaderMap;
7use schemars::JsonSchema;
8use serde::{Deserialize, Serialize};
9use serde_json::Value;
10use std::collections::HashMap;
11
12/// Authentication for GraphQL endpoints.
13#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
14#[serde(tag = "type", content = "config", rename_all = "snake_case")]
15pub enum GraphqlAuth {
16    /// No authentication.
17    None,
18    /// Bearer token in the Authorization header.
19    Bearer { token: String },
20    /// Custom headers (e.g. API keys, cookies).
21    Custom { headers: HashMap<String, String> },
22}
23
24/// Cursor-based pagination configuration for GraphQL.
25///
26/// Most GraphQL APIs use the Relay cursor specification with
27/// `pageInfo { hasNextPage, endCursor }`.
28#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
29pub struct GraphqlPagination {
30    /// JSONPath to the `hasNextPage` boolean in the response.
31    pub has_next_page_path: String,
32    /// JSONPath to the `endCursor` string in the response.
33    pub cursor_path: String,
34    /// Name of the cursor variable in the GraphQL query (default: `"after"`).
35    pub cursor_variable: String,
36    /// Name of the page size variable (default: `"first"`).
37    ///
38    /// The per-page record count itself comes from
39    /// [`GraphqlStreamConfig::batch_size`] — the variable named here is the
40    /// GraphQL variable that the `batch_size` value is injected into on each
41    /// request. The plain `batch_size = 0` sentinel omits the variable so the
42    /// upstream uses its own default page size.
43    pub page_size_variable: String,
44}
45
46impl Default for GraphqlPagination {
47    fn default() -> Self {
48        Self {
49            has_next_page_path: "$.data.*.pageInfo.hasNextPage".into(),
50            cursor_path: "$.data.*.pageInfo.endCursor".into(),
51            cursor_variable: "after".into(),
52            page_size_variable: "first".into(),
53        }
54    }
55}
56
57/// Discriminator for [`GraphqlOffsetPagination`]; serializes as `"Offset"`.
58///
59/// A dedicated single-variant enum (rather than a bare `String`) so the
60/// `type: Offset` marker is validated at config-load time and gives the
61/// untagged [`GraphqlPaginationSpec`] a reliable way to route an offset block.
62#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
63pub enum OffsetPaginationKind {
64    /// The offset pagination style.
65    Offset,
66}
67
68/// Offset-into-query-variable pagination (ShopifyQL and similar).
69///
70/// Increments an integer offset injected into a GraphQL variable and
71/// terminates on a **short page** (fewer than `page_size` records) — unlike
72/// cursor pagination, which follows a `pageInfo` boolean. Suited to APIs whose
73/// query language embeds `LIMIT … OFFSET …` (e.g. ShopifyQL): bake the limit
74/// into the query string and parameterize only the offset with `${…}`.
75///
76/// The offset starts at `0`, is sent as a JSON number in the `variables` map on
77/// every request, and advances by `page_size` after each page.
78#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
79#[serde(deny_unknown_fields)]
80pub struct GraphqlOffsetPagination {
81    /// Discriminator — must be `Offset`.
82    pub r#type: OffsetPaginationKind,
83    /// Name of the GraphQL variable that receives the current offset. It is
84    /// injected as a JSON number (starting at `0`, incremented by `page_size`
85    /// after each page). Reference it from the query string or as a variable
86    /// (e.g. `${q_offset}` for ShopifyQL).
87    pub offset_variable: String,
88    /// Records requested per page. Used both to advance the offset
89    /// (`offset += page_size`) and, with `stop_when_short`, to detect the final
90    /// page. Must be greater than `0`. This value is **not** injected into the
91    /// request — bake the limit into your query (`LIMIT 250 OFFSET ${q_offset}`).
92    pub page_size: usize,
93    /// Terminate when a page yields fewer than `page_size` records (default
94    /// `true`). When `false`, pagination continues until a fully empty page (or
95    /// `max_pages`) is reached.
96    #[serde(default = "default_true")]
97    pub stop_when_short: bool,
98    /// Substitute `${offset_variable}` occurrences in the **query string** with
99    /// the current offset before each request, instead of sending it as a
100    /// GraphQL variable. Required for query languages that embed the offset in a
101    /// string-literal argument — e.g. ShopifyQL's
102    /// `shopifyqlQuery(query: "… LIMIT 250 OFFSET ${q_offset}")`, where a GraphQL
103    /// variable cannot interpolate into a string literal (#569). Default
104    /// `false` (variable injection, the #550 behavior).
105    #[serde(default)]
106    pub substitute_in_query: bool,
107}
108
109fn default_true() -> bool {
110    true
111}
112
113/// Pagination style for the GraphQL source.
114///
115/// Deserialized **untagged** so the legacy cursor block (which carries no
116/// `type:` discriminator) keeps working unchanged, while the offset block is
117/// selected by its required `type: Offset` field. The two field sets are
118/// disjoint (cursor requires `cursor_path` / `has_next_page_path`; offset
119/// requires `type` / `offset_variable` / `page_size`), so serde routes each
120/// config to exactly one variant. The cursor variant is tried first.
121#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
122#[serde(untagged)]
123pub enum GraphqlPaginationSpec {
124    /// Relay-style cursor pagination (the original, `type`-less shape).
125    Cursor(GraphqlPagination),
126    /// Offset-into-variable pagination — selected by `type: Offset`.
127    Offset(GraphqlOffsetPagination),
128}
129
130/// Configuration for the GraphQL source.
131#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
132pub struct GraphqlStreamConfig {
133    /// GraphQL endpoint URL.
134    pub endpoint: String,
135    /// The GraphQL query string.
136    pub query: String,
137    /// Variables to pass with the query.
138    pub variables: Value,
139    /// Authentication: either inline (`{ type, config }`) or a `{ ref: <name> }`
140    /// pointer to a shared provider in the CLI's top-level `auth:` catalog.
141    pub auth: AuthSpec<GraphqlAuth>,
142    /// Additional request headers.
143    #[serde(skip, default)]
144    pub headers: HeaderMap,
145    /// JSONPath expression to extract records from the response.
146    pub records_path: Option<String>,
147    /// Pagination configuration. `None` for single-page queries. Accepts either
148    /// the Relay cursor block (no `type:`) or an offset block (`type: Offset`) —
149    /// see [`GraphqlPaginationSpec`].
150    pub pagination: Option<GraphqlPaginationSpec>,
151    /// Maximum number of pages to fetch.
152    pub max_pages: Option<usize>,
153    /// Records per emitted [`StreamPage`](faucet_core::StreamPage), and the
154    /// value injected as the GraphQL `first:` cursor argument (or whatever
155    /// variable name [`GraphqlPagination::page_size_variable`] specifies).
156    /// Defaults to [`DEFAULT_BATCH_SIZE`].
157    ///
158    /// `batch_size = 0` is the "no batching" sentinel: the page-size variable
159    /// is omitted from the request so the upstream uses its own default page
160    /// size, and the entire result set is emitted as a single page. If the
161    /// upstream schema requires a non-null `first:` argument this will
162    /// surface as `FaucetError::Config` at stream-time.
163    #[serde(default = "default_batch_size")]
164    pub batch_size: usize,
165    /// Optional client-certificate (mutual TLS) config. When set, the source
166    /// presents a client certificate on every request (data + inline auth token
167    /// request). Requires the crate's `mtls` feature.
168    #[serde(default)]
169    pub tls: Option<TlsClientConfig>,
170}
171
172fn default_batch_size() -> usize {
173    DEFAULT_BATCH_SIZE
174}
175
176impl GraphqlStreamConfig {
177    /// Create a new config with an endpoint and query.
178    pub fn new(endpoint: impl Into<String>, query: impl Into<String>) -> Self {
179        Self {
180            endpoint: endpoint.into(),
181            query: query.into(),
182            variables: Value::Object(Default::default()),
183            auth: AuthSpec::Inline(GraphqlAuth::None),
184            headers: HeaderMap::new(),
185            records_path: None,
186            pagination: None,
187            max_pages: None,
188            batch_size: DEFAULT_BATCH_SIZE,
189            tls: None,
190        }
191    }
192
193    /// Attach a mutual-TLS client identity (requires the `mtls` feature at build
194    /// time; otherwise [`GraphqlStream::try_new`](crate::GraphqlStream::try_new)
195    /// errors).
196    pub fn tls(mut self, tls: TlsClientConfig) -> Self {
197        self.tls = Some(tls);
198        self
199    }
200
201    /// Set the GraphQL variables.
202    pub fn variables(mut self, vars: Value) -> Self {
203        self.variables = vars;
204        self
205    }
206
207    /// Set the authentication method.
208    pub fn auth(mut self, auth: GraphqlAuth) -> Self {
209        self.auth = AuthSpec::Inline(auth);
210        self
211    }
212
213    /// Set additional headers.
214    pub fn headers(mut self, headers: HeaderMap) -> Self {
215        self.headers = headers;
216        self
217    }
218
219    /// Set the JSONPath expression for record extraction.
220    pub fn records_path(mut self, path: impl Into<String>) -> Self {
221        self.records_path = Some(path.into());
222        self
223    }
224
225    /// Enable cursor-based (Relay) pagination with the given configuration.
226    pub fn pagination(mut self, pagination: GraphqlPagination) -> Self {
227        self.pagination = Some(GraphqlPaginationSpec::Cursor(pagination));
228        self
229    }
230
231    /// Enable offset-into-variable pagination (ShopifyQL and similar).
232    pub fn offset_pagination(mut self, pagination: GraphqlOffsetPagination) -> Self {
233        self.pagination = Some(GraphqlPaginationSpec::Offset(pagination));
234        self
235    }
236
237    /// Set the maximum number of pages to fetch.
238    pub fn max_pages(mut self, max: usize) -> Self {
239        self.max_pages = Some(max);
240        self
241    }
242
243    /// Set the per-page record count for [`Source::stream_pages`](faucet_core::Source::stream_pages)
244    /// and the GraphQL `first:` cursor argument.
245    ///
246    /// Pass `0` to opt out of batching — the page-size variable is omitted
247    /// from the request so the upstream uses its own default page size, and
248    /// the response is emitted as a single [`StreamPage`](faucet_core::StreamPage).
249    pub fn with_batch_size(mut self, batch_size: usize) -> Self {
250        self.batch_size = batch_size;
251        self
252    }
253
254    /// Validate the config at load time so a bad config fails fast with a typed
255    /// [`FaucetError::Config`] instead of surfacing deep in a run: rejects an
256    /// out-of-range `batch_size` (`> MAX_BATCH_SIZE`) and an empty `endpoint` or
257    /// `query`.
258    pub fn validate(&self) -> Result<(), FaucetError> {
259        if self.endpoint.trim().is_empty() {
260            return Err(FaucetError::Config(
261                "GraphQL source requires a non-empty `endpoint`".into(),
262            ));
263        }
264        if self.query.trim().is_empty() {
265            return Err(FaucetError::Config(
266                "GraphQL source requires a non-empty `query`".into(),
267            ));
268        }
269        validate_batch_size(self.batch_size)?;
270        if let Some(GraphqlPaginationSpec::Offset(off)) = &self.pagination
271            && off.page_size == 0
272        {
273            return Err(FaucetError::Config(
274                "GraphQL offset pagination requires `page_size` > 0 \
275                 (a zero page size never advances the offset)"
276                    .into(),
277            ));
278        }
279        if let Some(tls) = &self.tls {
280            tls.validate()?;
281        }
282        Ok(())
283    }
284}
285
286#[cfg(test)]
287mod tests {
288    use super::*;
289    use serde_json::json;
290
291    #[test]
292    fn default_config() {
293        let config = GraphqlStreamConfig::new(
294            "https://api.example.com/graphql",
295            "query { users { id name } }",
296        );
297        assert_eq!(config.endpoint, "https://api.example.com/graphql");
298        assert!(config.records_path.is_none());
299        assert!(config.pagination.is_none());
300        assert!(config.max_pages.is_none());
301    }
302
303    #[test]
304    fn builder_methods() {
305        let config =
306            GraphqlStreamConfig::new("https://api.example.com/graphql", "query { users { id } }")
307                .variables(json!({"org": "acme"}))
308                .records_path("$.data.users.edges[*].node")
309                .max_pages(10)
310                .auth(GraphqlAuth::Bearer {
311                    token: "token".into(),
312                });
313        assert_eq!(config.variables["org"], "acme");
314        assert_eq!(config.records_path.unwrap(), "$.data.users.edges[*].node");
315        assert_eq!(config.max_pages, Some(10));
316    }
317
318    #[test]
319    fn default_pagination() {
320        let pag = GraphqlPagination::default();
321        assert_eq!(pag.cursor_variable, "after");
322        assert_eq!(pag.page_size_variable, "first");
323    }
324
325    #[test]
326    fn batch_size_defaults_to_default_batch_size() {
327        let config = GraphqlStreamConfig::new("https://api.example.com/graphql", "query { x }");
328        assert_eq!(config.batch_size, faucet_core::DEFAULT_BATCH_SIZE);
329    }
330
331    #[test]
332    fn with_batch_size_overrides_default() {
333        let config = GraphqlStreamConfig::new("https://api.example.com/graphql", "query { x }")
334            .with_batch_size(250);
335        assert_eq!(config.batch_size, 250);
336    }
337
338    #[test]
339    fn batch_size_zero_is_accepted_as_no_batching_sentinel() {
340        let config = GraphqlStreamConfig::new("https://api.example.com/graphql", "query { x }")
341            .with_batch_size(0);
342        assert_eq!(config.batch_size, 0);
343        assert!(faucet_core::validate_batch_size(config.batch_size).is_ok());
344    }
345
346    #[test]
347    fn batch_size_above_max_is_rejected_by_validate_batch_size() {
348        let config = GraphqlStreamConfig::new("https://api.example.com/graphql", "query { x }")
349            .with_batch_size(faucet_core::MAX_BATCH_SIZE + 1);
350        assert!(faucet_core::validate_batch_size(config.batch_size).is_err());
351    }
352
353    #[test]
354    fn batch_size_deserializes_from_json() {
355        let json = r#"{
356            "endpoint": "https://api.example.com/graphql",
357            "query": "query { x }",
358            "variables": {},
359            "auth": {"type": "none"},
360            "records_path": null,
361            "pagination": null,
362            "max_pages": null,
363            "batch_size": 500
364        }"#;
365        let config: GraphqlStreamConfig = serde_json::from_str(json).unwrap();
366        assert_eq!(config.batch_size, 500);
367    }
368
369    #[test]
370    fn validate_accepts_valid_config() {
371        assert!(
372            GraphqlStreamConfig::new("https://api.example.com/graphql", "query { x }")
373                .validate()
374                .is_ok()
375        );
376    }
377
378    #[test]
379    fn validate_rejects_oversized_batch_size() {
380        let config = GraphqlStreamConfig::new("https://api.example.com/graphql", "query { x }")
381            .with_batch_size(faucet_core::MAX_BATCH_SIZE + 1);
382        assert!(matches!(config.validate(), Err(FaucetError::Config(_))));
383    }
384
385    #[test]
386    fn validate_rejects_empty_endpoint() {
387        assert!(matches!(
388            GraphqlStreamConfig::new("  ", "query { x }").validate(),
389            Err(FaucetError::Config(_))
390        ));
391    }
392
393    #[test]
394    fn validate_rejects_empty_query() {
395        assert!(matches!(
396            GraphqlStreamConfig::new("https://api.example.com/graphql", "").validate(),
397            Err(FaucetError::Config(_))
398        ));
399    }
400
401    // ── Offset pagination ───────────────────────────────────────────────────
402
403    #[test]
404    fn offset_pagination_deserializes_with_type_tag() {
405        let json = r#"{
406            "type": "Offset",
407            "offset_variable": "q_offset",
408            "page_size": 250,
409            "stop_when_short": true
410        }"#;
411        let spec: GraphqlPaginationSpec = serde_json::from_str(json).unwrap();
412        match spec {
413            GraphqlPaginationSpec::Offset(off) => {
414                assert_eq!(off.r#type, OffsetPaginationKind::Offset);
415                assert_eq!(off.offset_variable, "q_offset");
416                assert_eq!(off.page_size, 250);
417                assert!(off.stop_when_short);
418            }
419            other => panic!("expected Offset variant, got {other:?}"),
420        }
421    }
422
423    #[test]
424    fn offset_pagination_stop_when_short_defaults_true() {
425        let json = r#"{ "type": "Offset", "offset_variable": "q_offset", "page_size": 100 }"#;
426        let spec: GraphqlPaginationSpec = serde_json::from_str(json).unwrap();
427        match spec {
428            GraphqlPaginationSpec::Offset(off) => assert!(
429                off.stop_when_short,
430                "stop_when_short must default to true when omitted"
431            ),
432            other => panic!("expected Offset variant, got {other:?}"),
433        }
434    }
435
436    #[test]
437    fn cursor_pagination_still_deserializes_without_type_tag() {
438        // Backward compatibility: the legacy cursor block has no `type:` field
439        // and must route to the Cursor variant untouched.
440        let json = r#"{
441            "has_next_page_path": "$.data.users.pageInfo.hasNextPage",
442            "cursor_path": "$.data.users.pageInfo.endCursor",
443            "cursor_variable": "after",
444            "page_size_variable": "first"
445        }"#;
446        let spec: GraphqlPaginationSpec = serde_json::from_str(json).unwrap();
447        match spec {
448            GraphqlPaginationSpec::Cursor(pag) => {
449                assert_eq!(pag.cursor_variable, "after");
450                assert_eq!(pag.page_size_variable, "first");
451            }
452            other => panic!("expected Cursor variant, got {other:?}"),
453        }
454    }
455
456    #[test]
457    fn full_config_with_offset_pagination_deserializes() {
458        let json = r#"{
459            "endpoint": "https://api.example.com/graphql",
460            "query": "{ orders(first: 250, offset: $q_offset) { id } }",
461            "variables": {},
462            "auth": {"type": "none"},
463            "records_path": "$.data.orders[*]",
464            "pagination": { "type": "Offset", "offset_variable": "q_offset", "page_size": 250 },
465            "max_pages": null,
466            "batch_size": 250
467        }"#;
468        let config: GraphqlStreamConfig = serde_json::from_str(json).unwrap();
469        assert!(matches!(
470            config.pagination,
471            Some(GraphqlPaginationSpec::Offset(_))
472        ));
473        assert!(config.validate().is_ok());
474    }
475
476    #[test]
477    fn offset_pagination_rejects_unknown_field() {
478        let json = r#"{
479            "type": "Offset",
480            "offset_variable": "q_offset",
481            "page_size": 250,
482            "bogus": true
483        }"#;
484        // `deny_unknown_fields` on the offset struct means an unknown field can't
485        // silently be ignored; the untagged enum then matches no variant.
486        assert!(serde_json::from_str::<GraphqlPaginationSpec>(json).is_err());
487    }
488
489    #[test]
490    fn offset_pagination_builder_wraps_offset_variant() {
491        let config = GraphqlStreamConfig::new("https://api.example.com/graphql", "query { x }")
492            .offset_pagination(GraphqlOffsetPagination {
493                r#type: OffsetPaginationKind::Offset,
494                offset_variable: "q_offset".into(),
495                page_size: 250,
496                stop_when_short: true,
497                substitute_in_query: false,
498            });
499        assert!(matches!(
500            config.pagination,
501            Some(GraphqlPaginationSpec::Offset(_))
502        ));
503    }
504
505    #[test]
506    fn validate_rejects_zero_page_size_offset() {
507        let config = GraphqlStreamConfig::new("https://api.example.com/graphql", "query { x }")
508            .offset_pagination(GraphqlOffsetPagination {
509                r#type: OffsetPaginationKind::Offset,
510                offset_variable: "q_offset".into(),
511                page_size: 0,
512                stop_when_short: true,
513                substitute_in_query: false,
514            });
515        assert!(matches!(config.validate(), Err(FaucetError::Config(_))));
516    }
517
518    #[test]
519    fn validate_accepts_nonzero_page_size_offset() {
520        let config = GraphqlStreamConfig::new("https://api.example.com/graphql", "query { x }")
521            .offset_pagination(GraphqlOffsetPagination {
522                r#type: OffsetPaginationKind::Offset,
523                offset_variable: "q_offset".into(),
524                page_size: 1,
525                stop_when_short: false,
526                substitute_in_query: false,
527            });
528        assert!(config.validate().is_ok());
529    }
530}
faucet_source_graphql/config.rs

faucet_source_graphql/
config.rs
