tawasuyu/brahman

1//! Static dependency graph derived from a `Manifest`.
2//!
3//! Two graphs in one structure:
4//!   - **Explicit graph** (`depends_on`): morphism-to-morphism edges declared
5//!     by the manifest author. Cycles here are an error — the graph is built
6//!     with cycle detection.
7//!   - **Data-flow indexes** (`reads`/`writes`): inverted indexes from
8//!     canonical entity tokens (`"Caja.saldo"` or `"Movimiento"`) to the
9//!     morphisms that read or write them. Self-loops in data flow are
10//!     legal (a morphism that reads a field and updates it is normal).
11//!
12//! Tokens are normalized at build time: a manifest's role-prefixed tokens
13//! (`"caja.saldo"`) become entity-prefixed (`"Caja.saldo"`) so cross-module
14//! queries work uniformly.
15
16use petgraph::algo::tarjan_scc;
17use petgraph::graph::{DiGraph, NodeIndex};
18use petgraph::visit::Topo;
19use std::collections::{HashMap, HashSet};
20use thiserror::Error;
21
22use crate::manifest::Manifest;
23
24#[derive(Debug, Error)]
25pub enum GraphError {
26    #[error("dependency cycle in `depends_on` involving morphisms {0:?}")]
27    Cycle(Vec<String>),
28    #[error("morphism `{0}` referenced in depends_on but not declared in this manifest")]
29    UnknownMorphism(String),
30}
31
32#[derive(Debug)]
33pub struct ManifestGraph {
34    /// Explicit `depends_on` graph. Edge `a -> b` means: morphism `b`
35    /// depends on `a`, so `a` must be available before `b`.
36    explicit: DiGraph<String, ()>,
37
38    /// Data-flow indexes. Token form: "Entity.field" or "Entity".
39    readers_of_token: HashMap<String, Vec<String>>,
40    writers_of_token: HashMap<String, Vec<String>>,
41
42    /// Per-morphism canonicalized token sets.
43    morphism_reads: HashMap<String, Vec<String>>,
44    morphism_writes: HashMap<String, Vec<String>>,
45}
46
47impl ManifestGraph {
48    pub fn build(manifest: &Manifest) -> Result<Self, GraphError> {
49        let explicit = build_explicit(manifest)?;
50        if let Some(cycle) = find_cycle(&explicit) {
51            return Err(GraphError::Cycle(cycle));
52        }
53        let (readers_of_token, writers_of_token, morphism_reads, morphism_writes) =
54            build_data_flow(manifest);
55        Ok(Self {
56            explicit,
57            readers_of_token,
58            writers_of_token,
59            morphism_reads,
60            morphism_writes,
61        })
62    }
63
64    /// Morphisms that read `token`. Token form: "Entity.field" or "Entity".
65    pub fn readers_of(&self, token: &str) -> &[String] {
66        self.readers_of_token
67            .get(token)
68            .map(|v| v.as_slice())
69            .unwrap_or(&[])
70    }
71
72    /// Morphisms that write `token`.
73    pub fn writers_of(&self, token: &str) -> &[String] {
74        self.writers_of_token
75            .get(token)
76            .map(|v| v.as_slice())
77            .unwrap_or(&[])
78    }
79
80    pub fn morphism_reads(&self, name: &str) -> &[String] {
81        self.morphism_reads
82            .get(name)
83            .map(|v| v.as_slice())
84            .unwrap_or(&[])
85    }
86
87    pub fn morphism_writes(&self, name: &str) -> &[String] {
88        self.morphism_writes
89            .get(name)
90            .map(|v| v.as_slice())
91            .unwrap_or(&[])
92    }
93
94    /// Morphisms whose `reads` overlap any of `name`'s `writes`. The
95    /// dirty-marking primitive: after `name` runs successfully, these are
96    /// the candidates whose derived state would be invalidated. The result
97    /// excludes `name` itself even if it reads what it writes.
98    pub fn affected_by(&self, name: &str) -> Vec<String> {
99        let writes = match self.morphism_writes.get(name) {
100            Some(w) => w,
101            None => return Vec::new(),
102        };
103        let mut affected: HashSet<String> = HashSet::new();
104        for token in writes {
105            if let Some(readers) = self.readers_of_token.get(token) {
106                for r in readers {
107                    if r != name {
108                        affected.insert(r.clone());
109                    }
110                }
111            }
112        }
113        let mut out: Vec<_> = affected.into_iter().collect();
114        out.sort();
115        out
116    }
117
118    /// Topological order of the explicit dependency graph. If `a` is in
119    /// `b.depends_on`, `a` precedes `b` in the result.
120    pub fn topological_order(&self) -> Vec<String> {
121        let mut topo = Topo::new(&self.explicit);
122        let mut out = Vec::new();
123        while let Some(idx) = topo.next(&self.explicit) {
124            out.push(self.explicit[idx].clone());
125        }
126        out
127    }
128}
129
130fn build_explicit(manifest: &Manifest) -> Result<DiGraph<String, ()>, GraphError> {
131    let mut graph = DiGraph::new();
132    let mut nodes: HashMap<String, NodeIndex> = HashMap::new();
133    for m in &manifest.morphisms {
134        let idx = graph.add_node(m.name.clone());
135        nodes.insert(m.name.clone(), idx);
136    }
137    for m in &manifest.morphisms {
138        let to = nodes[&m.name];
139        for dep in &m.depends_on {
140            let from = *nodes
141                .get(dep)
142                .ok_or_else(|| GraphError::UnknownMorphism(dep.clone()))?;
143            graph.add_edge(from, to, ());
144        }
145    }
146    Ok(graph)
147}
148
149/// Returns one cycle's nodes (sorted) if the graph has any. Self-loops
150/// are returned as `[name]`; multi-node SCCs as the SCC's nodes.
151fn find_cycle(graph: &DiGraph<String, ()>) -> Option<Vec<String>> {
152    for scc in tarjan_scc(graph) {
153        if scc.len() > 1 {
154            let mut names: Vec<String> = scc.iter().map(|i| graph[*i].clone()).collect();
155            names.sort();
156            return Some(names);
157        }
158        if scc.len() == 1 && graph.find_edge(scc[0], scc[0]).is_some() {
159            return Some(vec![graph[scc[0]].clone()]);
160        }
161    }
162    None
163}
164
165fn build_data_flow(
166    manifest: &Manifest,
167) -> (
168    HashMap<String, Vec<String>>,
169    HashMap<String, Vec<String>>,
170    HashMap<String, Vec<String>>,
171    HashMap<String, Vec<String>>,
172) {
173    let mut readers: HashMap<String, Vec<String>> = HashMap::new();
174    let mut writers: HashMap<String, Vec<String>> = HashMap::new();
175    let mut m_reads: HashMap<String, Vec<String>> = HashMap::new();
176    let mut m_writes: HashMap<String, Vec<String>> = HashMap::new();
177
178    for m in &manifest.morphisms {
179        let role_to_entity: HashMap<&str, &str> = m
180            .inputs
181            .iter()
182            .map(|i| (i.role.as_str(), i.entity.as_str()))
183            .collect();
184
185        // Dedupe per-morphism: `source.saldo` and `dest.saldo` both
186        // canonicalize to `Caja.saldo` — the morphism is one writer, not
187        // two.
188        let mut seen_reads: HashSet<String> = HashSet::new();
189        for r in &m.reads {
190            if let Some(token) = canonicalize_token(r, &role_to_entity) {
191                if seen_reads.insert(token.clone()) {
192                    readers.entry(token.clone()).or_default().push(m.name.clone());
193                    m_reads.entry(m.name.clone()).or_default().push(token);
194                }
195            }
196        }
197        let mut seen_writes: HashSet<String> = HashSet::new();
198        for w in &m.writes {
199            if let Some(token) = canonicalize_token(w, &role_to_entity) {
200                if seen_writes.insert(token.clone()) {
201                    writers.entry(token.clone()).or_default().push(m.name.clone());
202                    m_writes.entry(m.name.clone()).or_default().push(token);
203                }
204            }
205        }
206    }
207
208    (readers, writers, m_reads, m_writes)
209}
210
211/// "role.field" -> "Entity.field" via the inputs map; "Entity" -> "Entity".
212fn canonicalize_token(t: &str, roles: &HashMap<&str, &str>) -> Option<String> {
213    if let Some((role, field)) = t.split_once('.') {
214        roles
215            .get(role)
216            .map(|entity| format!("{}.{}", entity, field))
217    } else {
218        Some(t.to_string())
219    }
220}
221
222/// Tracks which morphisms have stale derived state because some morphism
223/// they read from was applied. Wire it next to your `execute_and_log`
224/// loop: after a successful run, call `mark_dirty_after(morphism, &graph)`;
225/// then any consumer (cached view, derived report, downstream pipeline)
226/// queries `is_dirty(name)` before using its cached output.
227///
228/// The tracker holds names only — it doesn't know what "recompute" means
229/// for any particular morphism. That's deliberate: the kernel exposes the
230/// invalidation primitive; what to do with the dirty set is the caller's.
231#[derive(Debug, Default, Clone)]
232pub struct DirtyTracker {
233    dirty: HashSet<String>,
234}
235
236impl DirtyTracker {
237    pub fn new() -> Self {
238        Self::default()
239    }
240
241    /// After `morphism_name` runs successfully, mark every morphism in
242    /// `graph.affected_by(morphism_name)` as dirty.
243    pub fn mark_dirty_after(&mut self, morphism_name: &str, graph: &ManifestGraph) {
244        for affected in graph.affected_by(morphism_name) {
245            self.dirty.insert(affected);
246        }
247    }
248
249    pub fn is_dirty(&self, morphism: &str) -> bool {
250        self.dirty.contains(morphism)
251    }
252
253    /// Sorted list of dirty morphisms. Stable order for UI/telemetry.
254    pub fn dirty(&self) -> Vec<String> {
255        let mut out: Vec<String> = self.dirty.iter().cloned().collect();
256        out.sort();
257        out
258    }
259
260    pub fn len(&self) -> usize {
261        self.dirty.len()
262    }
263
264    pub fn is_empty(&self) -> bool {
265        self.dirty.is_empty()
266    }
267
268    /// Clear the dirty flag for a specific morphism (call after the
269    /// caller has recomputed it).
270    pub fn clear(&mut self, morphism: &str) {
271        self.dirty.remove(morphism);
272    }
273
274    pub fn clear_all(&mut self) {
275        self.dirty.clear();
276    }
277}