oxirs_gql/
graphql_mesh.rs

1//! GraphQL Mesh Integration
2//!
3//! This module provides integration with GraphQL Mesh concepts for combining
4//! multiple data sources into a unified GraphQL API.
5//!
6//! ## Features
7//!
8//! - **Source Management**: Define and manage multiple data sources
9//! - **Transform Pipelines**: Apply transforms to source schemas
10//! - **Type Merging**: Merge types across sources
11//! - **Cross-Source Relationships**: Define relationships between sources
12//! - **Caching Layer**: Unified caching across sources
13
14use anyhow::{anyhow, Result};
15use serde::{Deserialize, Serialize};
16use std::collections::HashMap;
17use std::sync::Arc;
18use std::time::{Duration, SystemTime};
19use tokio::sync::RwLock;
20
21/// Data source type
22#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
23pub enum SourceType {
24    /// GraphQL endpoint
25    GraphQL,
26    /// REST API
27    REST,
28    /// OpenAPI specification
29    OpenAPI,
30    /// SOAP service
31    SOAP,
32    /// gRPC service
33    GRPC,
34    /// OData endpoint
35    OData,
36    /// Database (SQL)
37    Database,
38    /// JSON Schema
39    JsonSchema,
40    /// Custom source
41    Custom,
42}
43
44/// Data source configuration
45#[derive(Debug, Clone, Serialize, Deserialize)]
46pub struct DataSource {
47    /// Source identifier
48    pub id: String,
49    /// Source name
50    pub name: String,
51    /// Source type
52    pub source_type: SourceType,
53    /// Endpoint URL
54    pub endpoint: String,
55    /// Schema (SDL for GraphQL, spec for OpenAPI, etc.)
56    pub schema: Option<String>,
57    /// Authentication configuration
58    pub auth: Option<AuthConfig>,
59    /// Headers to include
60    pub headers: HashMap<String, String>,
61    /// Transforms to apply
62    pub transforms: Vec<Transform>,
63    /// Cache configuration
64    pub cache: Option<CacheConfig>,
65    /// Health check endpoint
66    pub health_check: Option<String>,
67    /// Is enabled
68    pub enabled: bool,
69    /// Metadata
70    pub metadata: HashMap<String, String>,
71}
72
73impl DataSource {
74    /// Create a new data source
75    pub fn new(id: &str, name: &str, source_type: SourceType, endpoint: &str) -> Self {
76        Self {
77            id: id.to_string(),
78            name: name.to_string(),
79            source_type,
80            endpoint: endpoint.to_string(),
81            schema: None,
82            auth: None,
83            headers: HashMap::new(),
84            transforms: Vec::new(),
85            cache: None,
86            health_check: None,
87            enabled: true,
88            metadata: HashMap::new(),
89        }
90    }
91
92    /// Add authentication
93    pub fn with_auth(mut self, auth: AuthConfig) -> Self {
94        self.auth = Some(auth);
95        self
96    }
97
98    /// Add header
99    pub fn with_header(mut self, key: &str, value: &str) -> Self {
100        self.headers.insert(key.to_string(), value.to_string());
101        self
102    }
103
104    /// Add transform
105    pub fn with_transform(mut self, transform: Transform) -> Self {
106        self.transforms.push(transform);
107        self
108    }
109
110    /// Set cache configuration
111    pub fn with_cache(mut self, cache: CacheConfig) -> Self {
112        self.cache = Some(cache);
113        self
114    }
115}
116
117/// Authentication configuration
118#[derive(Debug, Clone, Serialize, Deserialize)]
119pub enum AuthConfig {
120    /// No authentication
121    None,
122    /// Basic authentication
123    Basic { username: String, password: String },
124    /// Bearer token
125    Bearer { token: String },
126    /// API key
127    ApiKey { key: String, header: String },
128    /// OAuth2
129    OAuth2 {
130        client_id: String,
131        client_secret: String,
132        token_url: String,
133        scopes: Vec<String>,
134    },
135    /// Custom
136    Custom { config: HashMap<String, String> },
137}
138
139/// Transform type
140#[derive(Debug, Clone, Serialize, Deserialize)]
141pub enum Transform {
142    /// Prefix type names
143    Prefix { value: String },
144    /// Rename type
145    Rename { from: String, to: String },
146    /// Filter types
147    FilterTypes { include: Vec<String> },
148    /// Filter fields
149    FilterFields {
150        type_name: String,
151        include: Vec<String>,
152    },
153    /// Add field
154    AddField {
155        type_name: String,
156        field_name: String,
157        field_type: String,
158    },
159    /// Encapsulate with namespace
160    Encapsulate { namespace: String },
161    /// Snapshot (freeze schema)
162    Snapshot { name: String },
163    /// Custom transform
164    Custom {
165        name: String,
166        config: HashMap<String, String>,
167    },
168}
169
170/// Cache configuration
171#[derive(Debug, Clone, Serialize, Deserialize)]
172pub struct CacheConfig {
173    /// Cache TTL
174    pub ttl: Duration,
175    /// Max cache entries
176    pub max_entries: usize,
177    /// Cache key strategy
178    pub key_strategy: CacheKeyStrategy,
179    /// Invalidation events
180    pub invalidation_events: Vec<String>,
181}
182
183impl Default for CacheConfig {
184    fn default() -> Self {
185        Self {
186            ttl: Duration::from_secs(60),
187            max_entries: 1000,
188            key_strategy: CacheKeyStrategy::default(),
189            invalidation_events: Vec::new(),
190        }
191    }
192}
193
194/// Cache key strategy
195#[derive(Debug, Clone, Default, Serialize, Deserialize)]
196pub enum CacheKeyStrategy {
197    /// Full query hash
198    #[default]
199    QueryHash,
200    /// Field-level caching
201    FieldLevel,
202    /// Custom key
203    Custom { template: String },
204}
205
206/// Type merge configuration
207#[derive(Debug, Clone, Serialize, Deserialize)]
208pub struct TypeMergeConfig {
209    /// Type name
210    pub type_name: String,
211    /// Field to use as key
212    pub key_field: String,
213    /// Sources that provide this type
214    pub sources: Vec<String>,
215    /// Resolution strategy
216    pub resolution: MergeResolution,
217}
218
219/// Merge resolution strategy
220#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
221pub enum MergeResolution {
222    /// First source wins
223    First,
224    /// Last source wins
225    Last,
226    /// Merge all fields
227    MergeAll,
228    /// Fail on conflict
229    FailOnConflict,
230}
231
232/// Cross-source relationship
233#[derive(Debug, Clone, Serialize, Deserialize)]
234pub struct CrossSourceRelation {
235    /// Source type
236    pub source_type: String,
237    /// Source field
238    pub source_field: String,
239    /// Source ID
240    pub source_id: String,
241    /// Target type
242    pub target_type: String,
243    /// Target field for lookup
244    pub target_key_field: String,
245    /// Target source ID
246    pub target_source_id: String,
247    /// Relation type
248    pub relation_type: RelationType,
249}
250
251/// Relation type
252#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
253pub enum RelationType {
254    /// One to one
255    OneToOne,
256    /// One to many
257    OneToMany,
258    /// Many to one
259    ManyToOne,
260    /// Many to many
261    ManyToMany,
262}
263
264/// Mesh configuration
265#[derive(Debug, Clone, Serialize, Deserialize)]
266pub struct MeshConfig {
267    /// Name
268    pub name: String,
269    /// Data sources
270    pub sources: Vec<DataSource>,
271    /// Type merges
272    pub type_merges: Vec<TypeMergeConfig>,
273    /// Cross-source relations
274    pub relations: Vec<CrossSourceRelation>,
275    /// Global transforms
276    pub global_transforms: Vec<Transform>,
277    /// Enable introspection
278    pub enable_introspection: bool,
279    /// Serve config
280    pub serve: ServeConfig,
281}
282
283impl Default for MeshConfig {
284    fn default() -> Self {
285        Self {
286            name: "mesh".to_string(),
287            sources: Vec::new(),
288            type_merges: Vec::new(),
289            relations: Vec::new(),
290            global_transforms: Vec::new(),
291            enable_introspection: true,
292            serve: ServeConfig::default(),
293        }
294    }
295}
296
297/// Server configuration
298#[derive(Debug, Clone, Serialize, Deserialize)]
299pub struct ServeConfig {
300    /// Host
301    pub host: String,
302    /// Port
303    pub port: u16,
304    /// Enable playground
305    pub playground: bool,
306    /// CORS origins
307    pub cors: Vec<String>,
308}
309
310impl Default for ServeConfig {
311    fn default() -> Self {
312        Self {
313            host: "0.0.0.0".to_string(),
314            port: 4000,
315            playground: true,
316            cors: vec!["*".to_string()],
317        }
318    }
319}
320
321/// Source health status
322#[derive(Debug, Clone, Serialize, Deserialize)]
323pub struct SourceHealth {
324    /// Source ID
325    pub source_id: String,
326    /// Is healthy
327    pub healthy: bool,
328    /// Last check time
329    pub last_check: SystemTime,
330    /// Last error
331    pub last_error: Option<String>,
332    /// Latency (ms)
333    pub latency_ms: u64,
334}
335
336/// Internal state
337struct MeshState {
338    /// Config
339    config: MeshConfig,
340    /// Source health
341    source_health: HashMap<String, SourceHealth>,
342    /// Merged schema SDL
343    merged_schema: Option<String>,
344    /// Last merge time
345    last_merge: Option<SystemTime>,
346}
347
348impl MeshState {
349    fn new(config: MeshConfig) -> Self {
350        Self {
351            config,
352            source_health: HashMap::new(),
353            merged_schema: None,
354            last_merge: None,
355        }
356    }
357}
358
359/// GraphQL Mesh Manager
360///
361/// Manages multiple data sources and combines them into a unified GraphQL API.
362pub struct GraphQLMesh {
363    /// Internal state
364    state: Arc<RwLock<MeshState>>,
365}
366
367impl GraphQLMesh {
368    /// Create a new mesh from config
369    pub fn new(config: MeshConfig) -> Self {
370        Self {
371            state: Arc::new(RwLock::new(MeshState::new(config))),
372        }
373    }
374
375    /// Add a data source
376    pub async fn add_source(&self, source: DataSource) -> Result<()> {
377        let mut state = self.state.write().await;
378
379        if state.config.sources.iter().any(|s| s.id == source.id) {
380            return Err(anyhow!("Source '{}' already exists", source.id));
381        }
382
383        state.config.sources.push(source);
384        state.merged_schema = None; // Invalidate merged schema
385
386        Ok(())
387    }
388
389    /// Remove a data source
390    pub async fn remove_source(&self, source_id: &str) -> Result<()> {
391        let mut state = self.state.write().await;
392
393        let initial_len = state.config.sources.len();
394        state.config.sources.retain(|s| s.id != source_id);
395
396        if state.config.sources.len() == initial_len {
397            return Err(anyhow!("Source '{}' not found", source_id));
398        }
399
400        state.source_health.remove(source_id);
401        state.merged_schema = None;
402
403        Ok(())
404    }
405
406    /// Get all sources
407    pub async fn get_sources(&self) -> Vec<DataSource> {
408        let state = self.state.read().await;
409        state.config.sources.clone()
410    }
411
412    /// Get source by ID
413    pub async fn get_source(&self, source_id: &str) -> Option<DataSource> {
414        let state = self.state.read().await;
415        state
416            .config
417            .sources
418            .iter()
419            .find(|s| s.id == source_id)
420            .cloned()
421    }
422
423    /// Update source health
424    pub async fn update_source_health(
425        &self,
426        source_id: &str,
427        healthy: bool,
428        error: Option<String>,
429        latency_ms: u64,
430    ) {
431        let mut state = self.state.write().await;
432
433        state.source_health.insert(
434            source_id.to_string(),
435            SourceHealth {
436                source_id: source_id.to_string(),
437                healthy,
438                last_check: SystemTime::now(),
439                last_error: error,
440                latency_ms,
441            },
442        );
443    }
444
445    /// Get source health
446    pub async fn get_source_health(&self, source_id: &str) -> Option<SourceHealth> {
447        let state = self.state.read().await;
448        state.source_health.get(source_id).cloned()
449    }
450
451    /// Get all source health
452    pub async fn get_all_health(&self) -> Vec<SourceHealth> {
453        let state = self.state.read().await;
454        state.source_health.values().cloned().collect()
455    }
456
457    /// Add type merge configuration
458    pub async fn add_type_merge(&self, merge: TypeMergeConfig) {
459        let mut state = self.state.write().await;
460        state.config.type_merges.push(merge);
461        state.merged_schema = None;
462    }
463
464    /// Add cross-source relation
465    pub async fn add_relation(&self, relation: CrossSourceRelation) {
466        let mut state = self.state.write().await;
467        state.config.relations.push(relation);
468        state.merged_schema = None;
469    }
470
471    /// Build merged schema
472    pub async fn build_schema(&self) -> Result<String> {
473        let mut state = self.state.write().await;
474
475        let mut sdl = String::new();
476
477        // Build Query type from all sources
478        sdl.push_str("type Query {\n");
479
480        for source in &state.config.sources {
481            if !source.enabled {
482                continue;
483            }
484
485            // Apply prefix transform if present
486            let prefix = source
487                .transforms
488                .iter()
489                .find_map(|t| match t {
490                    Transform::Prefix { value } => Some(value.clone()),
491                    _ => None,
492                })
493                .unwrap_or_default();
494
495            sdl.push_str(&format!(
496                "  # Source: {} ({})\n",
497                source.name,
498                source.source_type.name()
499            ));
500            sdl.push_str(&format!(
501                "  {}health: Boolean @source(name: \"{}\")\n",
502                prefix.to_lowercase(),
503                source.id
504            ));
505        }
506
507        sdl.push_str("}\n\n");
508
509        // Add directive definition
510        sdl.push_str("directive @source(name: String!) on FIELD_DEFINITION\n");
511
512        state.merged_schema = Some(sdl.clone());
513        state.last_merge = Some(SystemTime::now());
514
515        Ok(sdl)
516    }
517
518    /// Get merged schema
519    pub async fn get_merged_schema(&self) -> Option<String> {
520        let state = self.state.read().await;
521        state.merged_schema.clone()
522    }
523
524    /// Get configuration
525    pub async fn get_config(&self) -> MeshConfig {
526        let state = self.state.read().await;
527        state.config.clone()
528    }
529
530    /// Validate configuration
531    pub async fn validate(&self) -> Vec<ValidationError> {
532        let state = self.state.read().await;
533        let mut errors = Vec::new();
534
535        // Check for duplicate source IDs
536        let mut seen_ids = std::collections::HashSet::new();
537        for source in &state.config.sources {
538            if !seen_ids.insert(&source.id) {
539                errors.push(ValidationError {
540                    path: format!("sources.{}", source.id),
541                    message: "Duplicate source ID".to_string(),
542                });
543            }
544        }
545
546        // Check type merges reference existing sources
547        for merge in &state.config.type_merges {
548            for source_id in &merge.sources {
549                if !state.config.sources.iter().any(|s| &s.id == source_id) {
550                    errors.push(ValidationError {
551                        path: format!("type_merges.{}.sources", merge.type_name),
552                        message: format!("Unknown source: {}", source_id),
553                    });
554                }
555            }
556        }
557
558        // Check relations reference existing sources
559        for relation in &state.config.relations {
560            if !state
561                .config
562                .sources
563                .iter()
564                .any(|s| s.id == relation.source_id)
565            {
566                errors.push(ValidationError {
567                    path: format!("relations.{}.source_id", relation.source_field),
568                    message: format!("Unknown source: {}", relation.source_id),
569                });
570            }
571            if !state
572                .config
573                .sources
574                .iter()
575                .any(|s| s.id == relation.target_source_id)
576            {
577                errors.push(ValidationError {
578                    path: format!("relations.{}.target_source_id", relation.source_field),
579                    message: format!("Unknown source: {}", relation.target_source_id),
580                });
581            }
582        }
583
584        errors
585    }
586
587    /// Export configuration as YAML
588    pub async fn export_yaml(&self) -> Result<String> {
589        let state = self.state.read().await;
590        let yaml = serde_json::to_string_pretty(&state.config)?;
591        Ok(yaml)
592    }
593}
594
595impl SourceType {
596    /// Get source type name
597    pub fn name(&self) -> &'static str {
598        match self {
599            SourceType::GraphQL => "GraphQL",
600            SourceType::REST => "REST",
601            SourceType::OpenAPI => "OpenAPI",
602            SourceType::SOAP => "SOAP",
603            SourceType::GRPC => "gRPC",
604            SourceType::OData => "OData",
605            SourceType::Database => "Database",
606            SourceType::JsonSchema => "JSON Schema",
607            SourceType::Custom => "Custom",
608        }
609    }
610}
611
612/// Validation error
613#[derive(Debug, Clone, Serialize, Deserialize)]
614pub struct ValidationError {
615    /// Error path
616    pub path: String,
617    /// Error message
618    pub message: String,
619}
620
621#[cfg(test)]
622mod tests {
623    use super::*;
624
625    #[tokio::test]
626    async fn test_mesh_creation() {
627        let config = MeshConfig::default();
628        let mesh = GraphQLMesh::new(config);
629
630        let sources = mesh.get_sources().await;
631        assert!(sources.is_empty());
632    }
633
634    #[tokio::test]
635    async fn test_add_source() {
636        let mesh = GraphQLMesh::new(MeshConfig::default());
637
638        let source = DataSource::new(
639            "users",
640            "Users API",
641            SourceType::GraphQL,
642            "https://api.example.com/graphql",
643        );
644
645        mesh.add_source(source).await.expect("should succeed");
646
647        let sources = mesh.get_sources().await;
648        assert_eq!(sources.len(), 1);
649        assert_eq!(sources[0].id, "users");
650    }
651
652    #[tokio::test]
653    async fn test_duplicate_source() {
654        let mesh = GraphQLMesh::new(MeshConfig::default());
655
656        let source1 = DataSource::new(
657            "api",
658            "API 1",
659            SourceType::GraphQL,
660            "https://api1.example.com",
661        );
662        let source2 = DataSource::new(
663            "api",
664            "API 2",
665            SourceType::GraphQL,
666            "https://api2.example.com",
667        );
668
669        mesh.add_source(source1).await.expect("should succeed");
670        let result = mesh.add_source(source2).await;
671
672        assert!(result.is_err());
673    }
674
675    #[tokio::test]
676    async fn test_remove_source() {
677        let mesh = GraphQLMesh::new(MeshConfig::default());
678
679        let source = DataSource::new(
680            "users",
681            "Users",
682            SourceType::GraphQL,
683            "https://api.example.com",
684        );
685        mesh.add_source(source).await.expect("should succeed");
686
687        mesh.remove_source("users").await.expect("should succeed");
688
689        let sources = mesh.get_sources().await;
690        assert!(sources.is_empty());
691    }
692
693    #[tokio::test]
694    async fn test_source_health() {
695        let mesh = GraphQLMesh::new(MeshConfig::default());
696
697        let source = DataSource::new("api", "API", SourceType::GraphQL, "https://api.example.com");
698        mesh.add_source(source).await.expect("should succeed");
699
700        mesh.update_source_health("api", true, None, 50).await;
701
702        let health = mesh.get_source_health("api").await.expect("should succeed");
703        assert!(health.healthy);
704        assert_eq!(health.latency_ms, 50);
705    }
706
707    #[tokio::test]
708    async fn test_build_schema() {
709        let mesh = GraphQLMesh::new(MeshConfig::default());
710
711        let source = DataSource::new(
712            "users",
713            "Users",
714            SourceType::GraphQL,
715            "https://api.example.com",
716        )
717        .with_transform(Transform::Prefix {
718            value: "Users_".to_string(),
719        });
720
721        mesh.add_source(source).await.expect("should succeed");
722
723        let schema = mesh.build_schema().await.expect("should succeed");
724        assert!(schema.contains("type Query"));
725        assert!(schema.contains("users_health"));
726    }
727
728    #[tokio::test]
729    async fn test_validation() {
730        let mesh = GraphQLMesh::new(MeshConfig::default());
731
732        mesh.add_type_merge(TypeMergeConfig {
733            type_name: "User".to_string(),
734            key_field: "id".to_string(),
735            sources: vec!["nonexistent".to_string()],
736            resolution: MergeResolution::MergeAll,
737        })
738        .await;
739
740        let errors = mesh.validate().await;
741        assert!(!errors.is_empty());
742    }
743
744    #[test]
745    fn test_data_source_builder() {
746        let source = DataSource::new("api", "API", SourceType::REST, "https://api.example.com")
747            .with_auth(AuthConfig::Bearer {
748                token: "secret".to_string(),
749            })
750            .with_header("X-Custom", "value")
751            .with_cache(CacheConfig::default());
752
753        assert!(source.auth.is_some());
754        assert!(source.headers.contains_key("X-Custom"));
755        assert!(source.cache.is_some());
756    }
757
758    #[test]
759    fn test_source_type_name() {
760        assert_eq!(SourceType::GraphQL.name(), "GraphQL");
761        assert_eq!(SourceType::REST.name(), "REST");
762        assert_eq!(SourceType::GRPC.name(), "gRPC");
763    }
764}
oxirs_gql/graphql_mesh.rs

oxirs_gql/
graphql_mesh.rs
