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}
99
100fn default_true() -> bool {
101    true
102}
103
104/// Pagination style for the GraphQL source.
105///
106/// Deserialized **untagged** so the legacy cursor block (which carries no
107/// `type:` discriminator) keeps working unchanged, while the offset block is
108/// selected by its required `type: Offset` field. The two field sets are
109/// disjoint (cursor requires `cursor_path` / `has_next_page_path`; offset
110/// requires `type` / `offset_variable` / `page_size`), so serde routes each
111/// config to exactly one variant. The cursor variant is tried first.
112#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
113#[serde(untagged)]
114pub enum GraphqlPaginationSpec {
115    /// Relay-style cursor pagination (the original, `type`-less shape).
116    Cursor(GraphqlPagination),
117    /// Offset-into-variable pagination — selected by `type: Offset`.
118    Offset(GraphqlOffsetPagination),
119}
120
121/// Configuration for the GraphQL source.
122#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
123pub struct GraphqlStreamConfig {
124    /// GraphQL endpoint URL.
125    pub endpoint: String,
126    /// The GraphQL query string.
127    pub query: String,
128    /// Variables to pass with the query.
129    pub variables: Value,
130    /// Authentication: either inline (`{ type, config }`) or a `{ ref: <name> }`
131    /// pointer to a shared provider in the CLI's top-level `auth:` catalog.
132    pub auth: AuthSpec<GraphqlAuth>,
133    /// Additional request headers.
134    #[serde(skip, default)]
135    pub headers: HeaderMap,
136    /// JSONPath expression to extract records from the response.
137    pub records_path: Option<String>,
138    /// Pagination configuration. `None` for single-page queries. Accepts either
139    /// the Relay cursor block (no `type:`) or an offset block (`type: Offset`) —
140    /// see [`GraphqlPaginationSpec`].
141    pub pagination: Option<GraphqlPaginationSpec>,
142    /// Maximum number of pages to fetch.
143    pub max_pages: Option<usize>,
144    /// Records per emitted [`StreamPage`](faucet_core::StreamPage), and the
145    /// value injected as the GraphQL `first:` cursor argument (or whatever
146    /// variable name [`GraphqlPagination::page_size_variable`] specifies).
147    /// Defaults to [`DEFAULT_BATCH_SIZE`].
148    ///
149    /// `batch_size = 0` is the "no batching" sentinel: the page-size variable
150    /// is omitted from the request so the upstream uses its own default page
151    /// size, and the entire result set is emitted as a single page. If the
152    /// upstream schema requires a non-null `first:` argument this will
153    /// surface as `FaucetError::Config` at stream-time.
154    #[serde(default = "default_batch_size")]
155    pub batch_size: usize,
156    /// Optional client-certificate (mutual TLS) config. When set, the source
157    /// presents a client certificate on every request (data + inline auth token
158    /// request). Requires the crate's `mtls` feature.
159    #[serde(default)]
160    pub tls: Option<TlsClientConfig>,
161}
162
163fn default_batch_size() -> usize {
164    DEFAULT_BATCH_SIZE
165}
166
167impl GraphqlStreamConfig {
168    /// Create a new config with an endpoint and query.
169    pub fn new(endpoint: impl Into<String>, query: impl Into<String>) -> Self {
170        Self {
171            endpoint: endpoint.into(),
172            query: query.into(),
173            variables: Value::Object(Default::default()),
174            auth: AuthSpec::Inline(GraphqlAuth::None),
175            headers: HeaderMap::new(),
176            records_path: None,
177            pagination: None,
178            max_pages: None,
179            batch_size: DEFAULT_BATCH_SIZE,
180            tls: None,
181        }
182    }
183
184    /// Attach a mutual-TLS client identity (requires the `mtls` feature at build
185    /// time; otherwise [`GraphqlStream::try_new`](crate::GraphqlStream::try_new)
186    /// errors).
187    pub fn tls(mut self, tls: TlsClientConfig) -> Self {
188        self.tls = Some(tls);
189        self
190    }
191
192    /// Set the GraphQL variables.
193    pub fn variables(mut self, vars: Value) -> Self {
194        self.variables = vars;
195        self
196    }
197
198    /// Set the authentication method.
199    pub fn auth(mut self, auth: GraphqlAuth) -> Self {
200        self.auth = AuthSpec::Inline(auth);
201        self
202    }
203
204    /// Set additional headers.
205    pub fn headers(mut self, headers: HeaderMap) -> Self {
206        self.headers = headers;
207        self
208    }
209
210    /// Set the JSONPath expression for record extraction.
211    pub fn records_path(mut self, path: impl Into<String>) -> Self {
212        self.records_path = Some(path.into());
213        self
214    }
215
216    /// Enable cursor-based (Relay) pagination with the given configuration.
217    pub fn pagination(mut self, pagination: GraphqlPagination) -> Self {
218        self.pagination = Some(GraphqlPaginationSpec::Cursor(pagination));
219        self
220    }
221
222    /// Enable offset-into-variable pagination (ShopifyQL and similar).
223    pub fn offset_pagination(mut self, pagination: GraphqlOffsetPagination) -> Self {
224        self.pagination = Some(GraphqlPaginationSpec::Offset(pagination));
225        self
226    }
227
228    /// Set the maximum number of pages to fetch.
229    pub fn max_pages(mut self, max: usize) -> Self {
230        self.max_pages = Some(max);
231        self
232    }
233
234    /// Set the per-page record count for [`Source::stream_pages`](faucet_core::Source::stream_pages)
235    /// and the GraphQL `first:` cursor argument.
236    ///
237    /// Pass `0` to opt out of batching — the page-size variable is omitted
238    /// from the request so the upstream uses its own default page size, and
239    /// the response is emitted as a single [`StreamPage`](faucet_core::StreamPage).
240    pub fn with_batch_size(mut self, batch_size: usize) -> Self {
241        self.batch_size = batch_size;
242        self
243    }
244
245    /// Validate the config at load time so a bad config fails fast with a typed
246    /// [`FaucetError::Config`] instead of surfacing deep in a run: rejects an
247    /// out-of-range `batch_size` (`> MAX_BATCH_SIZE`) and an empty `endpoint` or
248    /// `query`.
249    pub fn validate(&self) -> Result<(), FaucetError> {
250        if self.endpoint.trim().is_empty() {
251            return Err(FaucetError::Config(
252                "GraphQL source requires a non-empty `endpoint`".into(),
253            ));
254        }
255        if self.query.trim().is_empty() {
256            return Err(FaucetError::Config(
257                "GraphQL source requires a non-empty `query`".into(),
258            ));
259        }
260        validate_batch_size(self.batch_size)?;
261        if let Some(GraphqlPaginationSpec::Offset(off)) = &self.pagination
262            && off.page_size == 0
263        {
264            return Err(FaucetError::Config(
265                "GraphQL offset pagination requires `page_size` > 0 \
266                 (a zero page size never advances the offset)"
267                    .into(),
268            ));
269        }
270        if let Some(tls) = &self.tls {
271            tls.validate()?;
272        }
273        Ok(())
274    }
275}
276
277#[cfg(test)]
278mod tests {
279    use super::*;
280    use serde_json::json;
281
282    #[test]
283    fn default_config() {
284        let config = GraphqlStreamConfig::new(
285            "https://api.example.com/graphql",
286            "query { users { id name } }",
287        );
288        assert_eq!(config.endpoint, "https://api.example.com/graphql");
289        assert!(config.records_path.is_none());
290        assert!(config.pagination.is_none());
291        assert!(config.max_pages.is_none());
292    }
293
294    #[test]
295    fn builder_methods() {
296        let config =
297            GraphqlStreamConfig::new("https://api.example.com/graphql", "query { users { id } }")
298                .variables(json!({"org": "acme"}))
299                .records_path("$.data.users.edges[*].node")
300                .max_pages(10)
301                .auth(GraphqlAuth::Bearer {
302                    token: "token".into(),
303                });
304        assert_eq!(config.variables["org"], "acme");
305        assert_eq!(config.records_path.unwrap(), "$.data.users.edges[*].node");
306        assert_eq!(config.max_pages, Some(10));
307    }
308
309    #[test]
310    fn default_pagination() {
311        let pag = GraphqlPagination::default();
312        assert_eq!(pag.cursor_variable, "after");
313        assert_eq!(pag.page_size_variable, "first");
314    }
315
316    #[test]
317    fn batch_size_defaults_to_default_batch_size() {
318        let config = GraphqlStreamConfig::new("https://api.example.com/graphql", "query { x }");
319        assert_eq!(config.batch_size, faucet_core::DEFAULT_BATCH_SIZE);
320    }
321
322    #[test]
323    fn with_batch_size_overrides_default() {
324        let config = GraphqlStreamConfig::new("https://api.example.com/graphql", "query { x }")
325            .with_batch_size(250);
326        assert_eq!(config.batch_size, 250);
327    }
328
329    #[test]
330    fn batch_size_zero_is_accepted_as_no_batching_sentinel() {
331        let config = GraphqlStreamConfig::new("https://api.example.com/graphql", "query { x }")
332            .with_batch_size(0);
333        assert_eq!(config.batch_size, 0);
334        assert!(faucet_core::validate_batch_size(config.batch_size).is_ok());
335    }
336
337    #[test]
338    fn batch_size_above_max_is_rejected_by_validate_batch_size() {
339        let config = GraphqlStreamConfig::new("https://api.example.com/graphql", "query { x }")
340            .with_batch_size(faucet_core::MAX_BATCH_SIZE + 1);
341        assert!(faucet_core::validate_batch_size(config.batch_size).is_err());
342    }
343
344    #[test]
345    fn batch_size_deserializes_from_json() {
346        let json = r#"{
347            "endpoint": "https://api.example.com/graphql",
348            "query": "query { x }",
349            "variables": {},
350            "auth": {"type": "none"},
351            "records_path": null,
352            "pagination": null,
353            "max_pages": null,
354            "batch_size": 500
355        }"#;
356        let config: GraphqlStreamConfig = serde_json::from_str(json).unwrap();
357        assert_eq!(config.batch_size, 500);
358    }
359
360    #[test]
361    fn validate_accepts_valid_config() {
362        assert!(
363            GraphqlStreamConfig::new("https://api.example.com/graphql", "query { x }")
364                .validate()
365                .is_ok()
366        );
367    }
368
369    #[test]
370    fn validate_rejects_oversized_batch_size() {
371        let config = GraphqlStreamConfig::new("https://api.example.com/graphql", "query { x }")
372            .with_batch_size(faucet_core::MAX_BATCH_SIZE + 1);
373        assert!(matches!(config.validate(), Err(FaucetError::Config(_))));
374    }
375
376    #[test]
377    fn validate_rejects_empty_endpoint() {
378        assert!(matches!(
379            GraphqlStreamConfig::new("  ", "query { x }").validate(),
380            Err(FaucetError::Config(_))
381        ));
382    }
383
384    #[test]
385    fn validate_rejects_empty_query() {
386        assert!(matches!(
387            GraphqlStreamConfig::new("https://api.example.com/graphql", "").validate(),
388            Err(FaucetError::Config(_))
389        ));
390    }
391
392    // ── Offset pagination ───────────────────────────────────────────────────
393
394    #[test]
395    fn offset_pagination_deserializes_with_type_tag() {
396        let json = r#"{
397            "type": "Offset",
398            "offset_variable": "q_offset",
399            "page_size": 250,
400            "stop_when_short": true
401        }"#;
402        let spec: GraphqlPaginationSpec = serde_json::from_str(json).unwrap();
403        match spec {
404            GraphqlPaginationSpec::Offset(off) => {
405                assert_eq!(off.r#type, OffsetPaginationKind::Offset);
406                assert_eq!(off.offset_variable, "q_offset");
407                assert_eq!(off.page_size, 250);
408                assert!(off.stop_when_short);
409            }
410            other => panic!("expected Offset variant, got {other:?}"),
411        }
412    }
413
414    #[test]
415    fn offset_pagination_stop_when_short_defaults_true() {
416        let json = r#"{ "type": "Offset", "offset_variable": "q_offset", "page_size": 100 }"#;
417        let spec: GraphqlPaginationSpec = serde_json::from_str(json).unwrap();
418        match spec {
419            GraphqlPaginationSpec::Offset(off) => assert!(
420                off.stop_when_short,
421                "stop_when_short must default to true when omitted"
422            ),
423            other => panic!("expected Offset variant, got {other:?}"),
424        }
425    }
426
427    #[test]
428    fn cursor_pagination_still_deserializes_without_type_tag() {
429        // Backward compatibility: the legacy cursor block has no `type:` field
430        // and must route to the Cursor variant untouched.
431        let json = r#"{
432            "has_next_page_path": "$.data.users.pageInfo.hasNextPage",
433            "cursor_path": "$.data.users.pageInfo.endCursor",
434            "cursor_variable": "after",
435            "page_size_variable": "first"
436        }"#;
437        let spec: GraphqlPaginationSpec = serde_json::from_str(json).unwrap();
438        match spec {
439            GraphqlPaginationSpec::Cursor(pag) => {
440                assert_eq!(pag.cursor_variable, "after");
441                assert_eq!(pag.page_size_variable, "first");
442            }
443            other => panic!("expected Cursor variant, got {other:?}"),
444        }
445    }
446
447    #[test]
448    fn full_config_with_offset_pagination_deserializes() {
449        let json = r#"{
450            "endpoint": "https://api.example.com/graphql",
451            "query": "{ orders(first: 250, offset: $q_offset) { id } }",
452            "variables": {},
453            "auth": {"type": "none"},
454            "records_path": "$.data.orders[*]",
455            "pagination": { "type": "Offset", "offset_variable": "q_offset", "page_size": 250 },
456            "max_pages": null,
457            "batch_size": 250
458        }"#;
459        let config: GraphqlStreamConfig = serde_json::from_str(json).unwrap();
460        assert!(matches!(
461            config.pagination,
462            Some(GraphqlPaginationSpec::Offset(_))
463        ));
464        assert!(config.validate().is_ok());
465    }
466
467    #[test]
468    fn offset_pagination_rejects_unknown_field() {
469        let json = r#"{
470            "type": "Offset",
471            "offset_variable": "q_offset",
472            "page_size": 250,
473            "bogus": true
474        }"#;
475        // `deny_unknown_fields` on the offset struct means an unknown field can't
476        // silently be ignored; the untagged enum then matches no variant.
477        assert!(serde_json::from_str::<GraphqlPaginationSpec>(json).is_err());
478    }
479
480    #[test]
481    fn offset_pagination_builder_wraps_offset_variant() {
482        let config = GraphqlStreamConfig::new("https://api.example.com/graphql", "query { x }")
483            .offset_pagination(GraphqlOffsetPagination {
484                r#type: OffsetPaginationKind::Offset,
485                offset_variable: "q_offset".into(),
486                page_size: 250,
487                stop_when_short: true,
488            });
489        assert!(matches!(
490            config.pagination,
491            Some(GraphqlPaginationSpec::Offset(_))
492        ));
493    }
494
495    #[test]
496    fn validate_rejects_zero_page_size_offset() {
497        let config = GraphqlStreamConfig::new("https://api.example.com/graphql", "query { x }")
498            .offset_pagination(GraphqlOffsetPagination {
499                r#type: OffsetPaginationKind::Offset,
500                offset_variable: "q_offset".into(),
501                page_size: 0,
502                stop_when_short: true,
503            });
504        assert!(matches!(config.validate(), Err(FaucetError::Config(_))));
505    }
506
507    #[test]
508    fn validate_accepts_nonzero_page_size_offset() {
509        let config = GraphqlStreamConfig::new("https://api.example.com/graphql", "query { x }")
510            .offset_pagination(GraphqlOffsetPagination {
511                r#type: OffsetPaginationKind::Offset,
512                offset_variable: "q_offset".into(),
513                page_size: 1,
514                stop_when_short: false,
515            });
516        assert!(config.validate().is_ok());
517    }
518}
faucet_source_graphql/config.rs

faucet_source_graphql/
config.rs
