qualia_core_db/sparql_library/
sparql_federated.rs

1//! SPARQL 1.1 Federated Query (SERVICE)
2//!
3//! Implements federated query support with DID integration for CORS handling.
4
5use crate::sparql_ast::*;
6use crate::sparql_executor::QueryExecutor;
7use crate::sparql_parser;
8use crate::sparql_planner::QueryPlanner;
9use crate::NQuin;
10
11use std::collections::HashMap;
12
13/// SERVICE endpoint configuration
14#[repr(C)]
15#[derive(Debug, Clone, Copy)]
16pub struct ServiceEndpoint {
17    pub did: u64,          // DID identifier
18    pub endpoint_url: u64, // Hash of URL string
19    pub auth_method: u8,   // 0=none, 1=DID-LD, 2=JWT
20    pub timeout_ms: u32,
21}
22
23/// Federated query result
24#[repr(C)]
25#[derive(Debug, Clone, Copy)]
26pub struct FederatedResult {
27    pub endpoint_did: u64,
28    pub row_count: u16,
29    pub success: bool,
30}
31
32/// SPARQL Federated Query Handler — a DID-registry front over the URL-based
33/// [`FederatedQueryEngine`]. It maps a DID to a concrete endpoint URL string and
34/// dispatches remote SPARQL requests through the same real HTTP transport.
35pub struct FederatedQueryHandler<'a> {
36    pub local_quins: &'a [NQuin],
37    pub endpoints: [Option<ServiceEndpoint>; 16],
38    pub endpoint_count: u8,
39    pub cache_enabled: bool,
40    /// DID → concrete endpoint URL string (a `u64` hash cannot be un-hashed to a URL,
41    /// so the resolvable URL is stored alongside for real remote execution).
42    endpoint_urls: HashMap<u64, String>,
43}
44
45impl<'a> FederatedQueryHandler<'a> {
46    pub fn new(quins: &'a [NQuin]) -> Self {
47        Self {
48            local_quins: quins,
49            endpoints: [None; 16],
50            endpoint_count: 0,
51            cache_enabled: true,
52            endpoint_urls: HashMap::new(),
53        }
54    }
55
56    /// Associate a DID with a concrete endpoint URL so remote queries can actually be
57    /// dispatched over HTTP (the `ServiceEndpoint.endpoint_url` hash is not reversible).
58    pub fn set_endpoint_url(&mut self, did: u64, url: &str) {
59        self.endpoint_urls.insert(did, url.to_string());
60    }
61
62    /// Register a federated endpoint
63    pub fn register_endpoint(&mut self, endpoint: ServiceEndpoint) -> Result<u8, String> {
64        if self.endpoint_count >= 16 {
65            return Err("Endpoint overflow".to_string());
66        }
67        let idx = self.endpoint_count;
68        self.endpoints[idx as usize] = Some(endpoint);
69        self.endpoint_count += 1;
70        Ok(idx)
71    }
72
73    /// Resolve DID to endpoint URL
74    pub fn resolve_did(&self, did: u64) -> Result<u64, String> {
75        // In production, this would:
76        // 1. Resolve DID using DID resolver
77        // 2. Extract service endpoint from DID document
78        // 3. Return the endpoint URL hash
79
80        // Simplified: return DID as URL hash
81        for i in 0..self.endpoint_count as usize {
82            if let Some(endpoint) = self.endpoints[i] {
83                if endpoint.did == did {
84                    return Ok(endpoint.endpoint_url);
85                }
86            }
87        }
88
89        Err("DID not found in endpoint registry".to_string())
90    }
91
92    /// Execute a federated query against the endpoint registered for `did`, over the
93    /// real HTTP SPARQL 1.1 Protocol transport.
94    ///
95    /// Requires a resolvable endpoint URL registered via
96    /// [`set_endpoint_url`](Self::set_endpoint_url) (the `ServiceEndpoint.endpoint_url`
97    /// hash is not reversible to a URL). Authenticated (DID-LD/JWT) endpoints must be
98    /// signed by the identity layer; only the **no-auth** path is dispatched here — no
99    /// fabricated auth header. Network egress: makes a real outbound request.
100    pub fn execute_service(
101        &self,
102        did: u64,
103        query: &str,
104        _format: &str,
105    ) -> Result<FederatedResult, String> {
106        let endpoint = (0..self.endpoint_count as usize)
107            .filter_map(|i| self.endpoints[i])
108            .find(|e| e.did == did)
109            .ok_or("DID not found in endpoint registry")?;
110        if endpoint.auth_method != 0 {
111            return Err(
112                "authenticated (DID-LD/JWT) federation must be signed by the \
113                        identity layer; only no-auth remote endpoints are dispatched here"
114                    .to_string(),
115            );
116        }
117        let url = self.endpoint_urls.get(&did).ok_or(
118            "no resolvable endpoint URL registered for this DID (call set_endpoint_url); \
119             the endpoint_url hash is not reversible",
120        )?;
121        let body = fetch_remote_sparql(url, query, endpoint.timeout_ms, None)?;
122        let (_vars, rows, _lexicon) = parse_sparql_results_json(&body)?;
123        Ok(FederatedResult {
124            endpoint_did: did,
125            row_count: rows.len() as u16,
126            success: true,
127        })
128    }
129
130    /// Execute federated query with local data
131    pub fn execute_federated(
132        &self,
133        _service_did: u64,
134        _service_query: &str,
135        _local_pattern: PatternId,
136        _ctx: &SparqlQueryContext,
137    ) -> Result<Vec<BindingRow>, String> {
138        // Execute remote SERVICE query
139        // let _remote_result = self.execute_service(service_did, service_query, "json")?;
140
141        // Parse remote query to get variables
142        // let (_sparql_query, mut remote_ctx) = sparql_parser::parse_sparql(service_query)?;
143
144        // Execute local pattern
145        // let plan = QueryPlanner::from_pattern(local_pattern, ctx)?;
146        // let executor = QueryExecutor::new(self.local_quins);
147        // let local_results = executor.execute(&plan, ctx)?;
148
149        // Merge results (simplified: just return empty for now)
150        // In production, this would:
151        // 1. Get remote results from HTTP response
152        // 2. Join with local results on common variables
153        // 3. Return merged bindings
154
155        Ok(vec![])
156    }
157
158    /// Check CORS using DID-based authentication
159    pub fn check_cors_allowed(&self, did: u64, origin_did: u64) -> Result<bool, String> {
160        // In production, this would:
161        // 1. Resolve both DIDs
162        // 2. Check if origin_did is in the service endpoint's allowed origins list
163        // 3. Verify DID relationship (e.g., same controller, trusted relationship)
164
165        // Simplified: allow if DIDs are same
166        Ok(did == origin_did)
167    }
168
169    /// Generate CORS headers using DID
170    pub fn generate_cors_headers(&self, did: u64) -> Result<Vec<(u64, u64)>, String> {
171        // In production, this would generate proper CORS headers
172        // based on DID resolution and trust relationships
173
174        let access_control_origin: u64 = 0x41432D4F72696769; // "Access-Control-Origin" (truncated)
175        let access_control_methods: u64 = 0x41432D4D65746F64; // "Access-Control-Methods" (truncated)
176        let access_control_headers: u64 = 0x41432D4865616465; // "Access-Control-Headers" (truncated)
177
178        Ok(vec![
179            (access_control_origin, did),                     // Use DID as origin
180            (access_control_methods, 0x4745540000000000_u64), // "GET"
181            (access_control_headers, 0x436F6E74656E742D_u64), // "Content-Type" (truncated)
182        ])
183    }
184}
185
186impl<'a> Default for FederatedQueryHandler<'a> {
187    fn default() -> Self {
188        Self::new(&[])
189    }
190}
191
192// =============================================================================
193// FederatedQueryEngine — string-URL based federated query execution.
194//
195// This is a higher-level engine than the DID-hash `FederatedQueryHandler` above.
196// It works with human-readable endpoint URLs (`local:`, `qualia:`, `http://...`)
197// and is the entry point used by the SPARQL `SERVICE` clause evaluator.
198// =============================================================================
199
200/// A federated service endpoint identified by a readable URL string.
201///
202/// URL schemes:
203/// - `local:<graph>` / `qualia:<graph>` — execute against the local QualiaDB instance
204/// - `http://` / `https://`           — remote SPARQL endpoint (HTTP transport)
205#[derive(Debug, Clone)]
206pub struct FederatedService {
207    /// Endpoint URL, e.g. `local:default`, `qualia:graph1`, `https://db.example.org/sparql`.
208    pub endpoint: String,
209    /// 0 = none, 1 = DID-LD, 2 = JWT (forwarded to remote endpoints).
210    pub auth_method: u8,
211    /// Request timeout in milliseconds (remote endpoints only).
212    pub timeout_ms: u32,
213}
214
215impl FederatedService {
216    /// Create a new service endpoint with no auth and a 30s default timeout.
217    pub fn new(endpoint: &str) -> Self {
218        Self {
219            endpoint: endpoint.to_string(),
220            auth_method: 0,
221            timeout_ms: 30_000,
222        }
223    }
224
225    /// Returns true when the endpoint targets the local QualiaDB instance and
226    /// should be executed in-process rather than over HTTP.
227    pub fn is_local(&self) -> bool {
228        self.endpoint.starts_with("local:") || self.endpoint.starts_with("qualia:")
229    }
230
231    /// Returns the local graph identifier portion of a `local:`/`qualia:` endpoint,
232    /// or `None` for remote endpoints.
233    pub fn local_graph(&self) -> Option<&str> {
234        if let Some(rest) = self.endpoint.strip_prefix("local:") {
235            return Some(rest);
236        }
237        if let Some(rest) = self.endpoint.strip_prefix("qualia:") {
238            return Some(rest);
239        }
240        None
241    }
242}
243
244/// The outcome of executing a single query against a single federated service.
245#[derive(Debug, Clone)]
246pub struct FederatedQueryResult {
247    /// The endpoint URL the query was dispatched to.
248    pub service_endpoint: String,
249    /// The SPARQL query string that was executed.
250    pub query: String,
251    /// Result bindings — real rows for both local execution and successful remote
252    /// (HTTP) execution. Each slot holds a term hash resolvable via [`Self::lexicon`]
253    /// (remote) or the local graph lexicon (local).
254    pub results: Vec<BindingRow>,
255    /// Result variable names in slot order (populated for remote SPARQL-results
256    /// responses; empty for a bare local execution that carries its own ctx).
257    pub variables: Vec<String>,
258    /// Hash → text (+ lang/datatype) for the terms in `results`, so remote string
259    /// terms are resolvable in the local u64-term model. Empty for local execution
260    /// (whose terms resolve via the local graph lexicon).
261    pub lexicon: LiteralTable,
262    /// Whether execution succeeded.
263    pub success: bool,
264    /// Error message when `success` is false.
265    pub error: Option<String>,
266    /// For remote endpoints, the fully-constructed query URL that was fetched (kept
267    /// for provenance/debugging). `None` for local execution.
268    pub remote_query_url: Option<String>,
269}
270
271impl FederatedQueryResult {
272    /// Number of result rows.
273    pub fn row_count(&self) -> usize {
274        self.results.len()
275    }
276}
277
278/// A federated query: a SPARQL query to run across one or more services.
279#[derive(Debug, Clone)]
280pub struct FederatedQuery {
281    /// Services to dispatch the query to.
282    pub services: Vec<FederatedService>,
283    /// The SPARQL query string.
284    pub query: String,
285}
286
287impl FederatedQuery {
288    /// Create a new federated query targeting a single service.
289    pub fn new(query: &str) -> Self {
290        Self {
291            services: Vec::new(),
292            query: query.to_string(),
293        }
294    }
295
296    /// Add a service to the federated query.
297    pub fn with_service(mut self, service: FederatedService) -> Self {
298        self.services.push(service);
299        self
300    }
301}
302
303/// Federated query engine — dispatches queries across local and remote services.
304///
305/// Local services (`local:`/`qualia:`) are executed in-process via the standard
306/// SPARQL parser → planner → executor pipeline. Remote HTTP(S) services are executed
307/// over the **real SPARQL 1.1 Protocol** ([`fetch_remote_sparql`] +
308/// [`parse_sparql_results_json`]): a `application/sparql-query` POST, the
309/// `application/sparql-results+json` response parsed into binding rows + a resolvable
310/// lexicon. Remote execution is real network egress and runs only for endpoints the
311/// caller explicitly targets.
312pub struct FederatedQueryEngine<'a> {
313    /// Local Quins the engine executes local services against.
314    pub local_quins: &'a [NQuin],
315    /// Registered service endpoints keyed by endpoint URL.
316    pub services: HashMap<String, FederatedService>,
317    /// Whether [`initialize`](Self::initialize) has been called.
318    pub initialized: bool,
319}
320
321impl<'a> FederatedQueryEngine<'a> {
322    /// Create a new engine over a slice of local Quins.
323    pub fn new(quins: &'a [NQuin]) -> Self {
324        Self {
325            local_quins: quins,
326            services: HashMap::new(),
327            initialized: false,
328        }
329    }
330
331    /// Initialize the engine for use. Clears any stale state and marks the
332    /// engine ready. Calling this is required before executing federated
333    /// queries so that downstream code can rely on a deterministic start state.
334    pub fn initialize(&mut self) -> Result<(), String> {
335        // Reset to a clean, known-good state. Registered services are preserved
336        // (registration may happen before or after initialize), but the
337        // initialized flag is what gates execution.
338        self.initialized = true;
339        Ok(())
340    }
341
342    /// Register a federated service endpoint. Re-registering an existing
343    /// endpoint URL updates its configuration in place.
344    pub fn register_service(&mut self, service: FederatedService) {
345        self.services.insert(service.endpoint.clone(), service);
346    }
347
348    /// List the endpoint URLs of all registered services.
349    pub fn list_services(&self) -> Vec<String> {
350        let mut endpoints: Vec<String> = self.services.keys().cloned().collect();
351        endpoints.sort();
352        endpoints
353    }
354
355    /// Execute a query against a single federated service.
356    ///
357    /// - Local endpoints (`local:`/`qualia:`) are executed in-process and
358    ///   return real result bindings.
359    /// - Remote HTTP endpoints return a placeholder result containing the
360    ///   constructed query URL (`<endpoint>?query=<encoded>`). The HTTP fetch
361    ///   itself requires an async runtime which is not available in this
362    ///   synchronous module; the placeholder is structured so a transport layer
363    ///   can be plugged in later without changing the call sites.
364    pub fn execute_service(
365        &self,
366        service: &FederatedService,
367        query: &str,
368    ) -> Result<FederatedQueryResult, String> {
369        if service.is_local() {
370            // Execute locally through the standard pipeline.
371            let (sparql_query, ctx) = match sparql_parser::parse_sparql(query) {
372                Ok(parsed) => parsed,
373                Err(e) => {
374                    return Ok(Self::failed(
375                        service,
376                        query,
377                        format!("Parse error: {e}"),
378                        None,
379                    ))
380                }
381            };
382            let plan = match QueryPlanner::plan(&sparql_query, &ctx) {
383                Ok(plan) => plan,
384                Err(e) => {
385                    return Ok(Self::failed(
386                        service,
387                        query,
388                        format!("Planning error: {e}"),
389                        None,
390                    ))
391                }
392            };
393            let executor = QueryExecutor::new(self.local_quins);
394            match executor.execute(&plan, &ctx) {
395                Ok(results) => Ok(FederatedQueryResult {
396                    service_endpoint: service.endpoint.clone(),
397                    query: query.to_string(),
398                    results,
399                    variables: Vec::new(),
400                    lexicon: LiteralTable::new(),
401                    success: true,
402                    error: None,
403                    remote_query_url: None,
404                }),
405                Err(e) => Ok(Self::failed(
406                    service,
407                    query,
408                    format!("Execution error: {e}"),
409                    None,
410                )),
411            }
412        } else {
413            // Remote HTTP SPARQL endpoint: perform a real SPARQL 1.1 Protocol request and
414            // parse the `application/sparql-results+json` response into binding rows.
415            //
416            // Network egress: this makes a real outbound HTTP request, and runs only when
417            // a caller explicitly dispatches to a remote `FederatedService`. Authenticated
418            // (DID-LD/JWT) endpoints need the identity layer to sign the request; the
419            // no-auth path is supported here directly (no synthetic/fake auth header).
420            let query_url = build_remote_query_url(&service.endpoint, query);
421            match fetch_remote_sparql(&service.endpoint, query, service.timeout_ms, None) {
422                Ok(body) => match parse_sparql_results_json(&body) {
423                    Ok((variables, results, lexicon)) => Ok(FederatedQueryResult {
424                        service_endpoint: service.endpoint.clone(),
425                        query: query.to_string(),
426                        results,
427                        variables,
428                        lexicon,
429                        success: true,
430                        error: None,
431                        remote_query_url: Some(query_url),
432                    }),
433                    Err(e) => Ok(Self::failed(
434                        service,
435                        query,
436                        format!("Remote result parse error: {e}"),
437                        Some(query_url),
438                    )),
439                },
440                Err(e) => Ok(Self::failed(service, query, e, Some(query_url))),
441            }
442        }
443    }
444
445    /// Construct a failed [`FederatedQueryResult`] carrying an error message.
446    fn failed(
447        service: &FederatedService,
448        query: &str,
449        error: String,
450        remote_query_url: Option<String>,
451    ) -> FederatedQueryResult {
452        FederatedQueryResult {
453            service_endpoint: service.endpoint.clone(),
454            query: query.to_string(),
455            results: Vec::new(),
456            variables: Vec::new(),
457            lexicon: LiteralTable::new(),
458            success: false,
459            error: Some(error),
460            remote_query_url,
461        }
462    }
463
464    /// Execute a federated query across all of its services.
465    ///
466    /// Each service is dispatched via [`execute_service`](Self::execute_service)
467    /// and the per-service results are collected. Results are concatenated
468    /// (simple merge); full SPARQL join semantics across services are a
469    /// separate task.
470    pub fn execute_federated(
471        &self,
472        query: &FederatedQuery,
473    ) -> Result<Vec<FederatedQueryResult>, String> {
474        if !self.initialized {
475            return Err("FederatedQueryEngine not initialized".to_string());
476        }
477        if query.services.is_empty() {
478            return Err("Federated query has no services".to_string());
479        }
480
481        let mut results = Vec::with_capacity(query.services.len());
482        for service in &query.services {
483            // A per-service failure does not abort the whole federated query;
484            // the error is captured in the result so callers can decide.
485            let result = self.execute_service(service, &query.query)?;
486            results.push(result);
487        }
488        Ok(results)
489    }
490}
491
492impl<'a> Default for FederatedQueryEngine<'a> {
493    fn default() -> Self {
494        Self::new(&[])
495    }
496}
497
498/// Build the HTTP query URL for a remote SPARQL endpoint using simple
499/// percent-encoding of the query string. This is the URL an HTTP transport
500/// layer would GET/POST.
501fn build_remote_query_url(endpoint: &str, query: &str) -> String {
502    let encoded = percent_encode_query(query);
503    let separator = if endpoint.contains('?') { '&' } else { '?' };
504    format!("{}{}query={}", endpoint, separator, encoded)
505}
506
507/// Minimal percent-encoder for the SPARQL query parameter. Encodes characters
508/// that are not unreserved per RFC 3986. Kept dependency-free and synchronous.
509fn percent_encode_query(input: &str) -> String {
510    let mut out = String::with_capacity(input.len());
511    for &b in input.as_bytes() {
512        // Unreserved: A-Z a-z 0-9 - _ . ~
513        let unreserved = b.is_ascii_alphanumeric() || matches!(b, b'-' | b'_' | b'.' | b'~');
514        if unreserved {
515            out.push(b as char);
516        } else {
517            out.push('%');
518            out.push_str(&format!("{:02X}", b));
519        }
520    }
521    out
522}
523
524/// Execute a SPARQL query against a remote HTTP endpoint (SPARQL 1.1 Protocol) and
525/// return the raw `application/sparql-results+json` response body.
526///
527/// Uses a **blocking** HTTP client run on a dedicated thread, so it is safe to call from
528/// within a Tokio runtime (a blocking client cannot be created on an async worker
529/// thread). `timeout_ms` bounds the whole request. `auth`, when present, is sent as the
530/// `Authorization` header value.
531///
532/// **Network egress.** This performs a real outbound HTTP request to `endpoint`.
533#[cfg(not(target_arch = "wasm32"))]
534fn fetch_remote_sparql(
535    endpoint: &str,
536    query: &str,
537    timeout_ms: u32,
538    auth: Option<&str>,
539) -> Result<String, String> {
540    let endpoint = endpoint.to_string();
541    let query = query.to_string();
542    let auth = auth.map(|s| s.to_string());
543    std::thread::spawn(move || -> Result<String, String> {
544        let client = reqwest::blocking::Client::builder()
545            .timeout(std::time::Duration::from_millis(timeout_ms.max(1) as u64))
546            .build()
547            .map_err(|e| format!("http client build failed: {e}"))?;
548        let mut req = client
549            .post(&endpoint)
550            .header(reqwest::header::CONTENT_TYPE, "application/sparql-query")
551            .header(reqwest::header::ACCEPT, "application/sparql-results+json")
552            .body(query);
553        if let Some(a) = auth {
554            req = req.header(reqwest::header::AUTHORIZATION, a);
555        }
556        let resp = req
557            .send()
558            .map_err(|e| format!("remote request failed: {e}"))?;
559        let status = resp.status();
560        let text = resp
561            .text()
562            .map_err(|e| format!("reading remote response failed: {e}"))?;
563        if !status.is_success() {
564            let snippet: String = text.chars().take(200).collect();
565            return Err(format!("remote endpoint returned HTTP {status}: {snippet}"));
566        }
567        Ok(text)
568    })
569    .join()
570    .map_err(|_| "remote fetch thread panicked".to_string())?
571}
572
573/// wasm has no synchronous HTTP transport — remote federation is a native capability.
574#[cfg(target_arch = "wasm32")]
575fn fetch_remote_sparql(
576    _endpoint: &str,
577    _query: &str,
578    _timeout_ms: u32,
579    _auth: Option<&str>,
580) -> Result<String, String> {
581    Err("remote SPARQL federation is not available on the wasm target".to_string())
582}
583
584/// Parse an `application/sparql-results+json` document into `(variables, rows, lexicon)`:
585/// the result variable names in order, one [`BindingRow`] per solution (each slot a term
586/// hash), and a [`LiteralTable`] mapping those hashes back to text (with lang/datatype),
587/// so remote string terms are resolvable in the local u64-term model. Also handles the
588/// ASK form `{ "boolean": … }` (a single row with slot 0 = 1/0).
589fn parse_sparql_results_json(
590    body: &str,
591) -> Result<(Vec<String>, Vec<BindingRow>, LiteralTable), String> {
592    let v: serde_json::Value =
593        serde_json::from_str(body).map_err(|e| format!("invalid SPARQL results JSON: {e}"))?;
594    let mut lexicon = LiteralTable::new();
595
596    // ASK result → a single row, slot 0 = 1/0.
597    if let Some(b) = v.get("boolean").and_then(|b| b.as_bool()) {
598        let mut row = BindingRow::new();
599        row.slots[0] = Some(if b { 1 } else { 0 });
600        return Ok((Vec::new(), vec![row], lexicon));
601    }
602
603    let vars: Vec<String> = v
604        .get("head")
605        .and_then(|h| h.get("vars"))
606        .and_then(|a| a.as_array())
607        .map(|a| {
608            a.iter()
609                .filter_map(|x| x.as_str().map(String::from))
610                .collect()
611        })
612        .unwrap_or_default();
613
614    let bindings = v
615        .get("results")
616        .and_then(|r| r.get("bindings"))
617        .and_then(|b| b.as_array())
618        .cloned()
619        .unwrap_or_default();
620
621    let mut rows = Vec::with_capacity(bindings.len());
622    for binding in &bindings {
623        let mut row = BindingRow::new();
624        for (i, var) in vars.iter().enumerate() {
625            if i >= MAX_BINDINGS {
626                break; // fixed-capacity binding row (§6)
627            }
628            if let Some(term) = binding.get(var) {
629                let value = term.get("value").and_then(|x| x.as_str()).unwrap_or("");
630                let lang = term.get("xml:lang").and_then(|x| x.as_str());
631                let datatype = term.get("datatype").and_then(|x| x.as_str());
632                // Hash consistently with the local term model; intern text + any tags so
633                // the remote term round-trips through the resolver and STR/LANG/DATATYPE.
634                let hash = crate::sparql_ast::literal_term_hash(value, lang, datatype);
635                lexicon.intern_tagged(hash, value, lang, datatype);
636                row.slots[i] = Some(hash);
637            }
638        }
639        rows.push(row);
640    }
641    Ok((vars, rows, lexicon))
642}
643
644/// DID-based CORS helper
645pub struct DidCorsHelper;
646
647impl DidCorsHelper {
648    /// Verify a DID signature for a CORS preflight.
649    ///
650    /// The SPARQL federation layer holds **no key material** and receives only opaque
651    /// `u64` hashes (not the actual signature/challenge/public-key bytes), so it cannot
652    /// perform a real cryptographic verification here. It therefore **fails closed** with
653    /// a named error rather than returning a fabricated `true` (which would let any origin
654    /// pass). Real verification must go through the identity/key-vault layer, which holds
655    /// the resolved DID public key and the raw bytes.
656    pub fn verify_did_signature(
657        _did: u64,
658        _signature: u64,
659        _challenge: u64,
660    ) -> Result<bool, String> {
661        Err(
662            "DID signature verification must be performed by the identity/key-vault layer \
663             (the SPARQL federation layer holds no keys); refusing to fabricate a result"
664                .to_string(),
665        )
666    }
667
668    /// Generate DID-based challenge for CORS
669    pub fn generate_challenge(did: u64) -> u64 {
670        // In production, generate cryptographically secure challenge
671        did ^ 0xDEADBEEFCAFEBABE
672    }
673}
674
675#[cfg(test)]
676mod tests {
677    use super::*;
678
679    #[test]
680    fn test_federated_handler_creation() {
681        let quins = vec![];
682        let handler = FederatedQueryHandler::new(&quins);
683        assert_eq!(handler.endpoint_count, 0);
684    }
685
686    #[test]
687    fn test_register_endpoint() {
688        let quins = vec![];
689        let mut handler = FederatedQueryHandler::new(&quins);
690
691        let endpoint = ServiceEndpoint {
692            did: 1,
693            endpoint_url: 2,
694            auth_method: 0,
695            timeout_ms: 5000,
696        };
697
698        let result = handler.register_endpoint(endpoint);
699        assert!(result.is_ok());
700        assert_eq!(handler.endpoint_count, 1);
701    }
702
703    #[test]
704    fn test_resolve_did() {
705        let quins = vec![];
706        let mut handler = FederatedQueryHandler::new(&quins);
707
708        let endpoint = ServiceEndpoint {
709            did: 123,
710            endpoint_url: 456,
711            auth_method: 0,
712            timeout_ms: 5000,
713        };
714
715        handler.register_endpoint(endpoint).unwrap();
716        let url_hash = handler.resolve_did(123).unwrap();
717        assert_eq!(url_hash, 456);
718    }
719
720    #[test]
721    fn test_cors_check() {
722        let quins = vec![];
723        let handler = FederatedQueryHandler::new(&quins);
724
725        let allowed = handler.check_cors_allowed(123, 123).unwrap();
726        assert!(allowed);
727
728        let not_allowed = handler.check_cors_allowed(123, 456).unwrap();
729        assert!(!not_allowed);
730    }
731
732    #[test]
733    fn test_did_signature_verification_fails_closed() {
734        // The SPARQL layer holds no keys → verification is an honest error, NOT a
735        // fabricated `true` (which would let any origin pass a CORS preflight).
736        assert!(DidCorsHelper::verify_did_signature(123, 456, 789).is_err());
737    }
738
739    // ---- FederatedQueryEngine tests ----
740
741    #[test]
742    fn test_engine_creation() {
743        let quins = vec![];
744        let engine = FederatedQueryEngine::new(&quins);
745        assert!(!engine.initialized);
746        assert!(engine.services.is_empty());
747    }
748
749    #[test]
750    fn test_engine_initialize() {
751        let quins = vec![];
752        let mut engine = FederatedQueryEngine::new(&quins);
753        assert!(!engine.initialized);
754        engine.initialize().unwrap();
755        assert!(engine.initialized);
756    }
757
758    #[test]
759    fn test_register_and_list_services() {
760        let quins = vec![];
761        let mut engine = FederatedQueryEngine::new(&quins);
762        engine.register_service(FederatedService::new("local:default"));
763        engine.register_service(FederatedService::new("https://remote.example.org/sparql"));
764
765        let services = engine.list_services();
766        assert_eq!(services.len(), 2);
767        assert!(services.contains(&"local:default".to_string()));
768        assert!(services.contains(&"https://remote.example.org/sparql".to_string()));
769    }
770
771    #[test]
772    fn test_register_service_overwrites() {
773        let quins = vec![];
774        let mut engine = FederatedQueryEngine::new(&quins);
775        let mut svc = FederatedService::new("local:default");
776        svc.auth_method = 0;
777        engine.register_service(svc);
778        let mut svc2 = FederatedService::new("local:default");
779        svc2.auth_method = 2;
780        engine.register_service(svc2);
781
782        // Same endpoint URL → single entry, updated config.
783        assert_eq!(engine.list_services().len(), 1);
784        assert_eq!(engine.services.get("local:default").unwrap().auth_method, 2);
785    }
786
787    #[test]
788    fn test_service_is_local() {
789        assert!(FederatedService::new("local:default").is_local());
790        assert!(FederatedService::new("qualia:graph1").is_local());
791        assert!(!FederatedService::new("https://remote.example.org/sparql").is_local());
792        assert!(!FederatedService::new("http://remote.example.org/sparql").is_local());
793    }
794
795    #[test]
796    fn test_service_local_graph() {
797        assert_eq!(
798            FederatedService::new("local:default").local_graph(),
799            Some("default")
800        );
801        assert_eq!(
802            FederatedService::new("qualia:graph1").local_graph(),
803            Some("graph1")
804        );
805        assert_eq!(
806            FederatedService::new("https://remote.example.org/sparql").local_graph(),
807            None
808        );
809    }
810
811    #[test]
812    fn test_execute_service_local() {
813        let quins = vec![];
814        let engine = FederatedQueryEngine::new(&quins);
815        let service = FederatedService::new("local:default");
816        let result = engine
817            .execute_service(&service, "SELECT ?s WHERE { ?s ?p ?o }")
818            .unwrap();
819        assert_eq!(result.service_endpoint, "local:default");
820        assert!(
821            result.success,
822            "local execution should succeed: {:?}",
823            result.error
824        );
825        assert!(result.remote_query_url.is_none());
826        // No quins → no rows, but still a successful local execution.
827        assert_eq!(result.row_count(), 0);
828    }
829
830    #[test]
831    fn test_execute_service_local_parse_error() {
832        let quins = vec![];
833        let engine = FederatedQueryEngine::new(&quins);
834        let service = FederatedService::new("local:default");
835        let result = engine
836            .execute_service(&service, "this is not sparql")
837            .unwrap();
838        assert!(!result.success);
839        assert!(result.error.is_some());
840        assert_eq!(result.row_count(), 0);
841    }
842
843    #[test]
844    #[ignore = "makes a real (localhost) network connection — integration test"]
845    fn test_execute_service_remote_unreachable_fails_gracefully() {
846        let quins = vec![];
847        let engine = FederatedQueryEngine::new(&quins);
848        let mut service = FederatedService::new("http://127.0.0.1:1/sparql");
849        service.timeout_ms = 300;
850        let result = engine
851            .execute_service(&service, "SELECT ?s WHERE { ?s ?p ?o }")
852            .unwrap();
853        // A real fetch to an unreachable endpoint fails GRACEFULLY: success=false with a
854        // named error, and the constructed query URL is retained for provenance.
855        assert!(!result.success);
856        assert!(result.error.is_some());
857        assert!(result.remote_query_url.is_some());
858        assert_eq!(result.row_count(), 0);
859    }
860
861    #[test]
862    fn test_execute_federated_across_local_services() {
863        // Federation orchestration across multiple services, network-free (both local).
864        let quins = vec![];
865        let mut engine = FederatedQueryEngine::new(&quins);
866        engine.initialize().unwrap();
867
868        let query = FederatedQuery::new("SELECT ?s WHERE { ?s ?p ?o }")
869            .with_service(FederatedService::new("local:default"))
870            .with_service(FederatedService::new("qualia:graph1"));
871
872        let results = engine.execute_federated(&query).unwrap();
873        assert_eq!(results.len(), 2);
874        assert_eq!(results[0].service_endpoint, "local:default");
875        assert!(results[0].success);
876        assert!(results[0].remote_query_url.is_none());
877        assert_eq!(results[1].service_endpoint, "qualia:graph1");
878        assert!(results[1].success);
879    }
880
881    #[test]
882    fn test_parse_sparql_results_json_select() {
883        // A standard application/sparql-results+json SELECT response → real binding rows
884        // + a resolvable lexicon (incl. a language-tagged and a datatyped literal).
885        let body = r#"{
886          "head": { "vars": ["s", "name", "age"] },
887          "results": { "bindings": [
888            { "s": {"type":"uri","value":"http://ex/a"},
889              "name": {"type":"literal","xml:lang":"en","value":"Alice"},
890              "age": {"type":"literal","datatype":"http://www.w3.org/2001/XMLSchema#integer","value":"30"} }
891          ] }
892        }"#;
893        let (vars, rows, lexicon) = parse_sparql_results_json(body).unwrap();
894        assert_eq!(vars, vec!["s", "name", "age"]);
895        assert_eq!(rows.len(), 1);
896        let name_hash = rows[0].slots[1].unwrap();
897        assert_eq!(lexicon.resolve(name_hash), Some("Alice"));
898        assert_eq!(lexicon.lang(name_hash), Some("en"));
899        let age_hash = rows[0].slots[2].unwrap();
900        assert_eq!(
901            lexicon.datatype(age_hash),
902            Some("http://www.w3.org/2001/XMLSchema#integer")
903        );
904    }
905
906    #[test]
907    fn test_parse_sparql_results_json_ask() {
908        let (vars, rows, _lex) =
909            parse_sparql_results_json(r#"{ "head": {}, "boolean": true }"#).unwrap();
910        assert!(vars.is_empty());
911        assert_eq!(rows.len(), 1);
912        assert_eq!(rows[0].slots[0], Some(1));
913    }
914
915    #[test]
916    fn test_execute_federated_not_initialized() {
917        let quins = vec![];
918        let engine = FederatedQueryEngine::new(&quins);
919        let query = FederatedQuery::new("SELECT ?s WHERE { ?s ?p ?o }")
920            .with_service(FederatedService::new("local:default"));
921        let err = engine.execute_federated(&query).unwrap_err();
922        assert!(err.contains("not initialized"));
923    }
924
925    #[test]
926    fn test_execute_federated_no_services() {
927        let quins = vec![];
928        let mut engine = FederatedQueryEngine::new(&quins);
929        engine.initialize().unwrap();
930        let query = FederatedQuery::new("SELECT ?s WHERE { ?s ?p ?o }");
931        let err = engine.execute_federated(&query).unwrap_err();
932        assert!(err.contains("no services"));
933    }
934
935    #[test]
936    fn test_percent_encode_query() {
937        let encoded = percent_encode_query("SELECT ?s WHERE { ?s ?p ?o }");
938        assert!(!encoded.contains(' '));
939        assert!(encoded.contains("SELECT"));
940        // '?' should be encoded as %3F.
941        assert!(encoded.contains("%3F"));
942    }
943
944    #[test]
945    fn test_build_remote_query_url_joins_with_ampersand_when_query_present() {
946        // Endpoint already carrying a query string joins with '&', not a second '?'.
947        let url = build_remote_query_url("https://example.org/sparql?dataset=foo", "SELECT ?s");
948        assert!(url.contains("&query="));
949        assert!(!url.contains(' '));
950    }
951
952    #[test]
953    fn test_build_remote_query_url() {
954        let url = build_remote_query_url("https://example.org/sparql", "SELECT ?s");
955        assert!(url.starts_with("https://example.org/sparql?query="));
956        assert!(!url.contains(' '));
957    }
958}
qualia_core_db/sparql_library/sparql_federated.rs

qualia_core_db/sparql_library/
sparql_federated.rs
