1use std::collections::{BTreeMap, HashSet};
17use std::sync::{Arc, OnceLock};
18
19use core_query::cypher::{execute, is_write_tokens, lex, parse, plan, Params};
20use core_query::{expand, neighborhood, Dir, GraphView, ResultSet};
21use core_storage::fulltext::FulltextIndex;
22use core_storage::v8::seam::{ColumnsView, TopologyView};
23use core_storage::v8::MappedBase;
24use core_storage::wal::WalRecord;
25use core_storage::{
26 ColumnStore, Direction, EdgeProps, EdgePropsView, GraphError, IdMap, Interner, Result,
27 Topology, Value,
28};
29
30use crate::db::{EdgeInfo, NodeInfo};
31use crate::mask::NodeMask;
32use crate::roles::RoleDef;
33
34pub const FOLD_EVERY_K: usize = 64;
37
38pub struct CommitDelta {
40 pub records: Vec<WalRecord>,
42 pub derived_inserts: Vec<(u32, u32, u32)>,
44 pub derived_deletes: Vec<(u32, u32, u32)>,
46}
47
48#[derive(Clone)]
52pub struct FrozenOverlay {
53 pub ids: IdMap,
54 pub syms: Interner,
55 pub topo: Topology,
56 pub props: ColumnStore,
57 pub labels: Vec<u32>,
58 pub edge_props: EdgeProps,
59 pub roles: Option<Vec<RoleDef>>,
60 pub fulltext: FulltextIndex,
61}
62
63pub struct ReaderSnapshot {
72 pub frozen: Arc<FrozenOverlay>,
74 pub base: Option<Arc<MappedBase>>,
76 pub deltas: Vec<Arc<CommitDelta>>,
78 cache: OnceLock<std::result::Result<FrozenOverlay, String>>,
86}
87
88fn build_tv<'a>(topo: &'a Topology, base: &'a Option<Arc<MappedBase>>) -> TopologyView<'a> {
91 match base {
92 None => TopologyView::owned(topo),
93 Some(b) => {
94 let csr = b.topology().expect("base CSR CRC already verified at open");
95 TopologyView::with_base(topo, csr)
96 }
97 }
98}
99
100fn build_cv<'a>(props: &'a ColumnStore, base: &'a Option<Arc<MappedBase>>) -> ColumnsView<'a> {
101 match base {
102 None => ColumnsView::owned(props),
103 Some(b) => {
104 let cols = b
105 .columns()
106 .expect("base columns CRC already verified at open");
107 ColumnsView::with_base(props, cols)
108 }
109 }
110}
111
112fn build_epv<'a>(
113 edge_props: &'a EdgeProps,
114 base: &'a Option<Arc<MappedBase>>,
115) -> EdgePropsView<'a> {
116 match base {
117 None => EdgePropsView::owned(edge_props),
118 Some(b) => {
119 let archived = b
120 .edge_props_section()
121 .expect("base edge_props CRC already verified at open");
122 EdgePropsView::with_base(edge_props, archived)
123 }
124 }
125}
126
127fn make_view<'a>(
128 state: &'a FrozenOverlay,
129 base: &'a Option<Arc<MappedBase>>,
130 mask: Option<&'a HashSet<u32>>,
131) -> GraphView<'a> {
132 GraphView {
133 ids: &state.ids,
134 syms: &state.syms,
135 labels: &state.labels,
136 props: build_cv(&state.props, base),
137 topo: build_tv(&state.topo, base),
138 edge_props: build_epv(&state.edge_props, base),
139 mask,
140 }
141}
142
143#[allow(clippy::too_many_arguments)]
152fn apply_one(
153 ids: &mut IdMap,
154 syms: &mut Interner,
155 topo: &mut Topology,
156 props: &mut ColumnStore,
157 edge_props: &mut EdgeProps,
158 labels: &mut Vec<u32>,
159 fulltext: &mut FulltextIndex,
160 rec: &WalRecord,
161) -> Result<()> {
162 match rec {
163 WalRecord::Intern { id, text } => {
164 let got = syms.intern(text);
165 if got != *id {
166 return Err(GraphError::Corrupt {
167 detail: format!(
168 "mvcc delta intern mismatch for {text:?}: expected {id}, got {got}"
169 ),
170 });
171 }
172 }
173
174 WalRecord::InsertNodeId {
175 label,
176 key,
177 props: node_props,
178 } => {
179 let node_id = ids.try_insert(key)?;
180 if labels.len() <= node_id as usize {
181 labels.resize(node_id as usize + 1, u32::MAX);
182 }
183 labels[node_id as usize] = *label;
184 let label_str = syms
185 .resolve(*label)
186 .ok_or_else(|| GraphError::Corrupt {
187 detail: format!("mvcc delta: unknown label sym {label}"),
188 })?
189 .to_string();
190 for (field_sym, value) in node_props {
191 let field = syms
192 .resolve(*field_sym)
193 .ok_or_else(|| GraphError::Corrupt {
194 detail: format!("mvcc delta: unknown field sym {field_sym}"),
195 })?
196 .to_string();
197 props.set(node_id, &field, value.clone());
198 if fulltext.is_enabled(&label_str, &field) {
199 fulltext.add_tokens(node_id, &field, value);
200 }
201 }
202 }
203
204 WalRecord::InsertNode {
205 label,
206 key,
207 props: node_props,
208 } => {
209 let label_sym = syms.intern(label);
210 let node_id = ids.try_insert(key)?;
211 if labels.len() <= node_id as usize {
212 labels.resize(node_id as usize + 1, u32::MAX);
213 }
214 labels[node_id as usize] = label_sym;
215 for (field, value) in node_props {
216 props.set(node_id, field, value.clone());
217 if fulltext.is_enabled(label, field) {
218 fulltext.add_tokens(node_id, field, value);
219 }
220 }
221 }
222
223 WalRecord::SetPropId { id, field, value } => {
224 if let Some(field_str) = syms.resolve(*field).map(str::to_string) {
225 props.set(*id, &field_str, value.clone());
226 if let Some(&label_sym) = labels.get(*id as usize) {
227 if let Some(label_str) = syms.resolve(label_sym) {
228 if fulltext.is_enabled(label_str, &field_str) {
229 fulltext.add_tokens(*id, &field_str, value);
230 }
231 }
232 }
233 }
234 }
235
236 WalRecord::SetProp { key, field, value } => {
237 if let Some(node_id) = ids.get(key) {
238 props.set(node_id, field, value.clone());
239 if let Some(&label_sym) = labels.get(node_id as usize) {
240 if let Some(label_str) = syms.resolve(label_sym) {
241 if fulltext.is_enabled(label_str, field) {
242 fulltext.add_tokens(node_id, field, value);
243 }
244 }
245 }
246 }
247 }
248
249 WalRecord::RemoveProp { key, field } => {
250 if let Some(node_id) = ids.get(key) {
251 props.remove(node_id, field);
252 props.record_prop_tombstone(node_id, field);
257 fulltext.remove_node_field(node_id, field);
258 }
259 }
260
261 WalRecord::DeleteNode { key } => {
262 if let Some(node_id) = ids.delete(key) {
263 props.remove_all(node_id);
264 fulltext.remove_node(node_id);
265 if let Some(slot) = labels.get_mut(node_id as usize) {
267 *slot = u32::MAX;
268 }
269 let etypes: Vec<u32> = topo.etypes().collect();
273 let mut doomed = Vec::new();
274 for et in &etypes {
275 for &dst in topo.neighbors(*et, Direction::Out, node_id).as_ref() {
276 doomed.push((*et, node_id, dst));
277 }
278 for &src in topo.neighbors(*et, Direction::In, node_id).as_ref() {
279 doomed.push((*et, src, node_id));
280 }
281 }
282 for (et, s, d) in doomed {
283 topo.remove_edge(et, s, d);
284 edge_props.remove_edge(et, s, d);
285 }
286 }
287 }
288
289 WalRecord::InsertEdgeId { etype, src, dst } => {
290 topo.add_edge(*etype, *src, *dst);
291 }
292
293 WalRecord::InsertEdge {
294 edge_type,
295 src_key,
296 dst_key,
297 } => {
298 let etype = syms.intern(edge_type);
299 if let (Some(src), Some(dst)) = (ids.get(src_key), ids.get(dst_key)) {
300 topo.add_edge(etype, src, dst);
301 }
302 }
303
304 WalRecord::DeleteEdge {
305 edge_type,
306 src_key,
307 dst_key,
308 } => {
309 if let Some(etype) = syms.get(edge_type) {
310 if let (Some(src), Some(dst)) = (ids.get(src_key), ids.get(dst_key)) {
311 topo.remove_edge(etype, src, dst);
312 }
313 }
314 }
315
316 WalRecord::EnableFulltext { label, field } => {
317 fulltext.enable(label, field);
318 }
319
320 WalRecord::DisableFulltext { label, field } => {
321 fulltext.disable(label, field);
322 }
323
324 WalRecord::Batch(inner) => {
325 for r in inner {
326 apply_one(ids, syms, topo, props, edge_props, labels, fulltext, r)?;
327 }
328 }
329
330 WalRecord::CreateRule { .. }
333 | WalRecord::DeleteRule { .. }
334 | WalRecord::RebuildRule { .. }
335 | WalRecord::CreateView { .. }
336 | WalRecord::DeleteView { .. }
337 | WalRecord::DerivedEdgeAdded { .. }
341 | WalRecord::DerivedEdgeRetracted { .. } => {}
342
343 WalRecord::RenameNode { old_key, new_key } => {
344 if ids.get(old_key).is_some() {
347 ids.rename(old_key, new_key).map_err(|_| GraphError::Corrupt {
348 detail: format!("mvcc delta RenameNode {old_key}→{new_key} failed"),
349 })?;
350 }
351 }
352 }
353 Ok(())
354}
355
356fn mask_for_role_from(state: &FrozenOverlay, role: &str) -> Result<NodeMask> {
359 let roles = state.roles.as_ref().ok_or_else(|| GraphError::Corrupt {
360 detail: "roles.json was corrupt at open; fix the file and re-open".into(),
361 })?;
362 let def = roles
363 .iter()
364 .find(|r| r.name == role)
365 .ok_or_else(|| GraphError::KeyNotFound {
366 key: format!("role:{role}"),
367 })?;
368 let mut visible = HashSet::new();
369 for key in &def.keys {
370 if let Some(id) = state.ids.get(key) {
371 visible.insert(id);
372 }
373 }
374 for label_name in &def.labels {
375 if let Some(sym) = state.syms.get(label_name) {
376 for (i, &s) in state.labels.iter().enumerate() {
377 if s == sym {
378 visible.insert(i as u32);
379 }
380 }
381 }
382 }
383 Ok(NodeMask::from_ids(visible))
384}
385
386impl ReaderSnapshot {
387 fn materialize(&self) -> Result<FrozenOverlay> {
392 let mut w = (*self.frozen).clone();
393 for delta in &self.deltas {
394 for rec in &delta.records {
395 apply_one(
396 &mut w.ids,
397 &mut w.syms,
398 &mut w.topo,
399 &mut w.props,
400 &mut w.edge_props,
401 &mut w.labels,
402 &mut w.fulltext,
403 rec,
404 )?;
405 }
406 for &(etype, src, dst) in &delta.derived_inserts {
407 w.topo.add_edge(etype, src, dst);
408 }
409 for &(etype, src, dst) in &delta.derived_deletes {
410 w.topo.remove_edge(etype, src, dst);
411 }
412 }
413 if !self.deltas.is_empty() {
414 let cv = build_cv(&w.props, &self.base);
417 w.fulltext.rebuild_all(&w.ids, &w.labels, &w.syms, cv);
418 }
419 Ok(w)
420 }
421
422 pub(crate) fn new(
427 frozen: Arc<FrozenOverlay>,
428 base: Option<Arc<MappedBase>>,
429 deltas: Vec<Arc<CommitDelta>>,
430 ) -> Self {
431 Self {
432 frozen,
433 base,
434 deltas,
435 cache: OnceLock::new(),
436 }
437 }
438
439 fn effective(&self) -> Result<&FrozenOverlay> {
449 if self.deltas.is_empty() {
450 return Ok(&self.frozen);
451 }
452 let cached = self
453 .cache
454 .get_or_init(|| self.materialize().map_err(|e| e.to_string()));
455 cached
456 .as_ref()
457 .map_err(|e| GraphError::Corrupt { detail: e.clone() })
458 }
459
460 pub fn mask_for_role(&self, role: &str) -> Result<NodeMask> {
468 mask_for_role_from(self.effective()?, role)
469 }
470
471 pub fn resolve_key(&self, key: &str) -> Option<u32> {
476 self.effective().ok()?.ids.get(key)
477 }
478
479 pub fn query(&self, cypher: &str, params: &BTreeMap<String, Value>) -> Result<ResultSet> {
481 let tokens = lex(cypher).map_err(|e| GraphError::QueryError {
482 detail: format!("lex: {e}"),
483 })?;
484 let ast = parse(&tokens).map_err(|e| GraphError::QueryError {
485 detail: format!("parse: {e}"),
486 })?;
487 let ops = plan(&ast).map_err(|e| GraphError::QueryError {
488 detail: format!("plan: {e}"),
489 })?;
490 let state = self.effective()?;
491 let view = make_view(state, &self.base, None);
492 execute(&view, &ops, &Params(params)).map_err(|e| GraphError::QueryError {
493 detail: format!("execute: {e}"),
494 })
495 }
496
497 pub fn query_masked(
501 &self,
502 cypher: &str,
503 params: &BTreeMap<String, Value>,
504 mask: &NodeMask,
505 ) -> Result<ResultSet> {
506 let tokens = lex(cypher).map_err(|e| GraphError::QueryError {
507 detail: format!("lex: {e}"),
508 })?;
509 if is_write_tokens(&tokens) {
510 return Err(GraphError::QueryError {
511 detail: "masked queries are read-only".into(),
512 });
513 }
514 let ast = parse(&tokens).map_err(|e| GraphError::QueryError {
515 detail: format!("parse: {e}"),
516 })?;
517 let ops = plan(&ast).map_err(|e| GraphError::QueryError {
518 detail: format!("plan: {e}"),
519 })?;
520 let state = self.effective()?;
521 let view = make_view(state, &self.base, Some(&mask.visible));
522 execute(&view, &ops, &Params(params)).map_err(|e| GraphError::QueryError {
523 detail: format!("execute: {e}"),
524 })
525 }
526
527 pub fn node_info(&self, key: &str) -> Option<NodeInfo> {
529 node_info_from(key, self.effective().ok()?, &self.base)
530 }
531
532 pub fn node_edges(&self, key: &str) -> Result<Vec<EdgeInfo>> {
537 node_edges_from(key, self.effective()?, &self.base)
538 }
539
540 pub fn neighborhood_masked(
545 &self,
546 key: &str,
547 depth: u32,
548 edge_types: Option<&[&str]>,
549 dir: Dir,
550 mask: &NodeMask,
551 ) -> Option<ResultSet> {
552 neighborhood_masked_from(
553 key,
554 self.effective().ok()?,
555 &self.base,
556 depth,
557 edge_types,
558 dir,
559 mask,
560 )
561 }
562}
563
564fn node_info_from(
567 key: &str,
568 state: &FrozenOverlay,
569 base: &Option<Arc<MappedBase>>,
570) -> Option<NodeInfo> {
571 let id = state.ids.get(key)?;
572 let label_sym = *state.labels.get(id as usize)?;
573 if label_sym == u32::MAX {
574 return None;
575 }
576 let label = state.syms.resolve(label_sym)?.to_string();
577 let cv = build_cv(&state.props, base);
578 let mut props = BTreeMap::new();
579 for field in cv.field_names() {
580 if let Some(vr) = cv.get(id, &field) {
581 props.insert(field, vr.into_value());
582 }
583 }
584 Some(NodeInfo {
585 key: key.to_string(),
586 label,
587 props,
588 })
589}
590
591fn node_edges_from(
592 key: &str,
593 state: &FrozenOverlay,
594 base: &Option<Arc<MappedBase>>,
595) -> Result<Vec<EdgeInfo>> {
596 let id = state
597 .ids
598 .get(key)
599 .ok_or_else(|| GraphError::KeyNotFound { key: key.into() })?;
600 let tv = build_tv(&state.topo, base);
601 let mut edges = Vec::new();
602 for etype in tv.etypes() {
603 let edge_type = state
604 .syms
605 .resolve(etype)
606 .ok_or_else(|| GraphError::Corrupt {
607 detail: format!("reader: topology etype {etype} not in interner"),
608 })?
609 .to_string();
610 for dir in [Direction::Out, Direction::In] {
611 for &nbr in tv.neighbors(etype, dir, id).as_ref() {
612 let (src_key, dst_key) = match dir {
613 Direction::Out => (
614 key.to_string(),
615 state
616 .ids
617 .key_of(nbr)
618 .ok_or_else(|| GraphError::Corrupt {
619 detail: format!("topology id {nbr} has no key"),
620 })?
621 .to_string(),
622 ),
623 Direction::In => (
624 state
625 .ids
626 .key_of(nbr)
627 .ok_or_else(|| GraphError::Corrupt {
628 detail: format!("topology id {nbr} has no key"),
629 })?
630 .to_string(),
631 key.to_string(),
632 ),
633 };
634 edges.push(EdgeInfo {
635 edge_type: edge_type.clone(),
636 src_key,
637 dst_key,
638 derived: false,
639 });
640 }
641 }
642 }
643 edges.sort_by(|a, b| {
644 a.edge_type
645 .cmp(&b.edge_type)
646 .then(a.src_key.cmp(&b.src_key))
647 .then(a.dst_key.cmp(&b.dst_key))
648 });
649 edges.dedup();
650 Ok(edges)
651}
652
653fn neighborhood_masked_from(
654 key: &str,
655 state: &FrozenOverlay,
656 base: &Option<Arc<MappedBase>>,
657 depth: u32,
658 edge_types: Option<&[&str]>,
659 dir: Dir,
660 mask: &NodeMask,
661) -> Option<ResultSet> {
662 let start_id = state.ids.get(key)?;
663 let view = make_view(state, base, Some(&mask.visible));
664 let resolved: Option<Vec<u32>> = edge_types.map(|names| {
665 names
666 .iter()
667 .filter_map(|name| view.syms.get(name))
668 .collect()
669 });
670 let nb = neighborhood(&view, start_id, depth, resolved.as_deref(), dir);
671 let mut rs = ResultSet::new(vec!["key".into(), "label".into(), "depth".into()]);
672 let mut visited: Vec<(u32, u32)> = Vec::with_capacity(nb.nodes.len() + 1);
674 visited.push((start_id, 0));
675 for (nid, d) in &nb.nodes {
676 let k = view.key_of(*nid);
677 let lbl = view
678 .label_of(*nid)
679 .expect("real nodes always have a label; u32::MAX sentinel cannot occur");
680 rs.push_row(vec![
681 Some(Value::Str(k.to_string())),
682 Some(Value::Str(lbl.to_string())),
683 Some(Value::Int(*d as i64)),
684 ]);
685 visited.push((*nid, *d));
686 }
687 if mask.mode() == crate::mask::MaskMode::Stub {
695 let raw_view = make_view(state, base, None);
696 let mut seen: HashSet<u32> = visited.iter().map(|(id, _)| *id).collect();
697 for (node_id, node_depth) in &visited {
698 if *node_depth >= depth {
699 continue;
700 }
701 for e in expand(&raw_view, *node_id, resolved.as_deref(), dir) {
702 let nbr = if e.src == *node_id { e.dst } else { e.src };
703 if !mask.contains_id(nbr) && seen.insert(nbr) {
704 if let Some(k) = state.ids.key_of(nbr) {
705 rs.push_row(vec![
706 Some(Value::Str(k.to_string())),
707 None,
708 Some(Value::Int((*node_depth + 1) as i64)),
709 ]);
710 }
711 }
712 }
713 }
714 }
715 Some(rs)
716}