use oxigraph::model::{GraphName, Literal, NamedNode, NamedOrBlankNode, Quad, Term};
use oxigraph::sparql::{QueryResults, SparqlEvaluator};
use oxigraph::store::Store;
use std::collections::HashMap;
use crate::runtime::control_plane::receipts::{Blake3Hash, CryptographicReceipt};
use super::clock::{hash_query_results, preprocess_query, ReplayClock, ReplayEntropy};
fn find_matching_graph_name(store: &Store, hash_or_uri: &str) -> Option<GraphName> {
if hash_or_uri == "default" || hash_or_uri.is_empty() {
return Some(GraphName::DefaultGraph);
}
for g in store.named_graphs().flatten() {
let g_str = g.to_string();
if g_str == hash_or_uri || g_str.contains(hash_or_uri) {
match g {
NamedOrBlankNode::NamedNode(ref n) => return Some(GraphName::NamedNode(n.clone())),
NamedOrBlankNode::BlankNode(ref b) => return Some(GraphName::BlankNode(b.clone())),
}
}
if let NamedOrBlankNode::NamedNode(ref n) = g {
if n.as_str().ends_with(hash_or_uri) {
return Some(GraphName::NamedNode(n.clone()));
}
}
}
None
}
#[derive(Debug, Clone)]
pub struct ReplayDetail {
pub receipt: String,
pub graph_hash: String,
pub query_hash: String,
pub expected_hash: String,
pub actual_hash: String,
pub success: bool,
pub error: Option<String>,
}
#[derive(Debug, Clone)]
pub struct ReplaySummary {
pub total_replayed: usize,
pub total_success: usize,
pub total_mismatch: usize,
pub details: Vec<ReplayDetail>,
}
#[derive(Debug, Clone)]
pub struct QueryConsequenceReplayVerifier {
pub clock: ReplayClock,
pub entropy: ReplayEntropy,
pub query_registry: HashMap<String, String>,
}
impl QueryConsequenceReplayVerifier {
pub fn new(clock_time_ms: u64, entropy_seed: u64) -> Self {
let mut query_registry = HashMap::new();
query_registry.insert(
crate::runtime::sha256(super::super::invariants::QUERY_INVARIANT_1.as_bytes()),
super::super::invariants::QUERY_INVARIANT_1.to_string(),
);
query_registry.insert(
crate::runtime::sha256(super::super::invariants::QUERY_INVARIANT_2.as_bytes()),
super::super::invariants::QUERY_INVARIANT_2.to_string(),
);
query_registry.insert(
crate::runtime::sha256(super::super::invariants::QUERY_INVARIANT_3.as_bytes()),
super::super::invariants::QUERY_INVARIANT_3.to_string(),
);
query_registry.insert(
crate::runtime::sha256(super::super::invariants::QUERY_INVARIANT_4.as_bytes()),
super::super::invariants::QUERY_INVARIANT_4.to_string(),
);
query_registry.insert(
crate::runtime::sha256(super::super::invariants::QUERY_INVARIANT_5.as_bytes()),
super::super::invariants::QUERY_INVARIANT_5.to_string(),
);
Self {
clock: ReplayClock::new(clock_time_ms),
entropy: ReplayEntropy::new(entropy_seed),
query_registry,
}
}
pub fn register_query(&mut self, hash: String, query_str: String) {
self.query_registry.insert(hash, query_str);
}
pub fn verify_receipt(
&mut self,
store: &Store,
receipt_node: &NamedOrBlankNode,
expected_result_hash: &str,
query_hash: &str,
graph_hash: &str,
) -> Result<String, String> {
let isolated_store = Store::new().map_err(|e| e.to_string())?;
let graph_name =
find_matching_graph_name(store, graph_hash).unwrap_or(GraphName::DefaultGraph);
for quad_res in store.quads_for_pattern(None, None, None, Some(graph_name.as_ref())) {
let quad = quad_res.map_err(|e| e.to_string())?;
isolated_store
.insert(&Quad::new(
quad.subject,
quad.predicate,
quad.object,
GraphName::DefaultGraph,
))
.map_err(|e| e.to_string())?;
}
let mut db_query_str = None;
let db_lookup_q = format!(
"PREFIX max: <urn:lsp-max:core:> SELECT ?qStr WHERE {{ ?q a max:Query ; max:queryHash \"{}\" ; max:queryString ?qStr . }}",
query_hash
);
if let Ok(evaluator) = SparqlEvaluator::new().parse_query(&db_lookup_q) {
if let Ok(QueryResults::Solutions(mut solutions)) = evaluator.on_store(store).execute()
{
if let Some(Ok(sol)) = solutions.next() {
if let Some(Term::Literal(lit)) = sol.get("qStr") {
db_query_str = Some(lit.value().to_string());
}
}
}
}
let query_str = db_query_str
.or_else(|| self.query_registry.get(query_hash).cloned())
.ok_or_else(|| {
format!(
"Query string not registered or found for hash: {}",
query_hash
)
})?;
let preprocessed = preprocess_query(&query_str, &self.clock, &mut self.entropy);
let parsed = SparqlEvaluator::new()
.parse_query(&preprocessed)
.map_err(|e| e.to_string())?;
let results = parsed
.on_store(&isolated_store)
.execute()
.map_err(|e| e.to_string())?;
let actual_hash = hash_query_results(results)?;
if actual_hash != expected_result_hash {
return Err(format!(
"Replay mismatch on receipt {}: expected result hash '{}', got '{}'",
receipt_node, expected_result_hash, actual_hash
));
}
Ok(actual_hash)
}
pub fn verify_all(&mut self, store: &Store) -> Result<ReplaySummary, String> {
let query_receipts = "
PREFIX max: <urn:lsp-max:core:>
SELECT ?receipt ?resultHash ?queryHash ?graphHash ?g WHERE {
{
?receipt a max:Receipt ;
max:resultHash ?resultHash ;
max:queryHash ?queryHash ;
max:graphHash ?graphHash .
}
UNION
{
GRAPH ?g {
?receipt a max:Receipt ;
max:resultHash ?resultHash ;
max:queryHash ?queryHash ;
max:graphHash ?graphHash .
}
}
}
";
let parsed = SparqlEvaluator::new()
.parse_query(query_receipts)
.map_err(|e| e.to_string())?;
let query_results = parsed
.on_store(store)
.execute()
.map_err(|e| e.to_string())?;
let mut receipts_to_verify = Vec::new();
if let QueryResults::Solutions(solutions) = query_results {
for sol_res in solutions {
let sol = sol_res.map_err(|e| e.to_string())?;
if let (Some(rcpt), Some(res_hash), Some(q_hash), Some(g_hash)) = (
sol.get("receipt"),
sol.get("resultHash"),
sol.get("queryHash"),
sol.get("graphHash"),
) {
let graph_ctx = sol.get("g").cloned();
receipts_to_verify.push((
rcpt.clone(),
res_hash.clone(),
q_hash.clone(),
g_hash.clone(),
graph_ctx,
));
}
}
}
let mut details = Vec::new();
let mut total_success = 0;
let mut total_mismatch = 0;
for (rcpt_term, res_hash_term, q_hash_term, g_hash_term, graph_ctx) in receipts_to_verify {
let rcpt_node = match rcpt_term {
Term::NamedNode(ref n) => NamedOrBlankNode::NamedNode(n.clone()),
Term::BlankNode(ref b) => NamedOrBlankNode::BlankNode(b.clone()),
_ => continue,
};
let expected_hash = match res_hash_term {
Term::Literal(ref lit) => lit.value().to_string(),
_ => continue,
};
let query_hash = match q_hash_term {
Term::Literal(ref lit) => lit.value().to_string(),
_ => continue,
};
let graph_hash = match g_hash_term {
Term::Literal(ref lit) => lit.value().to_string(),
_ => continue,
};
let target_graph = match graph_ctx {
Some(Term::NamedNode(ref n)) => GraphName::NamedNode(n.clone()),
Some(Term::BlankNode(ref b)) => GraphName::NamedNode(
NamedNode::new(format!("urn:blank:{}", b.as_str())).unwrap(),
),
_ => GraphName::DefaultGraph,
};
let res =
self.verify_receipt(store, &rcpt_node, &expected_hash, &query_hash, &graph_hash);
let (success, actual_hash, err_msg) = match res {
Ok(h) => (true, h, None),
Err(e) => (false, expected_hash.clone(), Some(e)),
};
if success {
total_success += 1;
} else {
total_mismatch += 1;
}
let replay_uuid = self.entropy.next_uuid();
let replay_uri = format!("urn:lsp-max:replay:{}", replay_uuid);
let replay_node = NamedOrBlankNode::NamedNode(NamedNode::new(&replay_uri).unwrap());
store
.insert(&Quad::new(
replay_node.clone(),
NamedNode::new("http://www.w3.org/1999/02/22-rdf-syntax-ns#type").unwrap(),
Term::NamedNode(NamedNode::new("urn:lsp-max:core:Replay").unwrap()),
target_graph.clone(),
))
.map_err(|e| e.to_string())?;
store
.insert(&Quad::new(
replay_node.clone(),
NamedNode::new("urn:lsp-max:core:queryHash").unwrap(),
Term::Literal(Literal::new_simple_literal(query_hash.clone())),
target_graph.clone(),
))
.map_err(|e| e.to_string())?;
store
.insert(&Quad::new(
replay_node.clone(),
NamedNode::new("urn:lsp-max:core:graphHash").unwrap(),
Term::Literal(Literal::new_simple_literal(graph_hash.clone())),
target_graph.clone(),
))
.map_err(|e| e.to_string())?;
store
.insert(&Quad::new(
replay_node.clone(),
NamedNode::new("urn:lsp-max:core:resultHash").unwrap(),
Term::Literal(Literal::new_simple_literal(actual_hash.clone())),
target_graph.clone(),
))
.map_err(|e| e.to_string())?;
details.push(ReplayDetail {
receipt: rcpt_node.to_string(),
graph_hash,
query_hash,
expected_hash: expected_hash.clone(),
actual_hash,
success,
error: err_msg,
});
}
Ok(ReplaySummary {
total_replayed: details.len(),
total_success,
total_mismatch,
details,
})
}
}
pub struct ReplayVerifier;
impl ReplayVerifier {
pub fn verify_replay(
chain: &[CryptographicReceipt],
verifying_key: &ed25519_dalek::VerifyingKey,
expected_genesis_hash: &Blake3Hash,
) -> Result<(), String> {
if chain.is_empty() {
return Err("Chain is empty".to_string());
}
super::super::receipts::verify_receipt_chain(chain, verifying_key, expected_genesis_hash)
.map_err(|e| format!("Cryptographic validation failed: {}", e))?;
for receipt in chain.iter() {
let recomputed_hash = match receipt.sequence {
0 => Blake3Hash([0u8; 32]),
_ => receipt.consequence_hash, };
if recomputed_hash.0 != receipt.consequence_hash.0 {
return Err(format!(
"State consequence hash mismatch at sequence {}: expected {:?}, got {:?}",
receipt.sequence, receipt.consequence_hash, recomputed_hash
));
}
}
Ok(())
}
}
pub fn verify_replay(
chain: &[CryptographicReceipt],
verifying_key: &ed25519_dalek::VerifyingKey,
expected_genesis_hash: &Blake3Hash,
) -> Result<(), String> {
ReplayVerifier::verify_replay(chain, verifying_key, expected_genesis_hash)
}