code_system_graph_core/
event_graph.rs

1use code_system_graph_model::{
2    Edge, EdgeId, EdgeKind, EpistemicStatus, Evidence, EvidenceId, Node, NodeId, NodeKind, Provenance, RepoId, stable_id
3};
4
5use crate::{EventBroker, EventDocument, EventObservation, EventRole};
6
7/// Graph facts assembled from workspace-wide event declarations and source observations.
8#[derive(Debug, Clone, Default)]
9pub struct EventGraphFacts {
10    /// Event channels, schemas, source artifacts, and repository nodes.
11    pub nodes: Vec<Node>,
12    /// Publish, subscribe, delivery, schema, and dead-letter relationships.
13    pub edges: Vec<Edge>,
14    /// Direct contract and source-call evidence.
15    pub evidence: Vec<Evidence>,
16}
17
18struct EventEndpoint {
19    repo_id: RepoId,
20    source_path: String,
21    observation: EventObservation,
22    evidence: Evidence,
23}
24
25/// Converts event documents into deterministic, ambiguity-safe graph facts.
26///
27/// Source observations without a namespace link to a declared broker/channel only when exactly
28/// one namespace candidate exists. Generic source APIs link to a broker-specific declaration only
29/// when the channel has one unambiguous declaration across the workspace.
30#[must_use]
31pub fn event_documents_to_graph(
32    inputs: &[(&RepoId, &str, &str, &EventDocument)],
33) -> EventGraphFacts {
34    let mut result = EventGraphFacts::default();
35    let mut declarations = Vec::new();
36    let mut endpoints = Vec::new();
37    for (repo_id, source_path, content_hash, document) in inputs {
38        let repository = repository_node(repo_id);
39        let artifact = artifact_node(repo_id, source_path);
40        let artifact_evidence =
41            event_evidence(repo_id, source_path, content_hash, None, "event artifact");
42        result.edges.push(edge(
43            &repository.id,
44            &artifact.id,
45            EdgeKind::Contains,
46            vec![artifact_evidence.id.clone()],
47        ));
48        result.nodes.extend([repository, artifact]);
49        result.evidence.push(artifact_evidence);
50        for observation in &document.observations {
51            if observation.channel.is_none() || observation.incomplete {
52                continue;
53            }
54            let evidence = event_evidence(
55                repo_id,
56                source_path,
57                content_hash,
58                Some(observation),
59                "event boundary",
60            );
61            let endpoint = EventEndpoint {
62                repo_id: (*repo_id).clone(),
63                source_path: (*source_path).to_owned(),
64                observation: observation.clone(),
65                evidence,
66            };
67            if observation.language.is_none() || observation.role == EventRole::Declaration {
68                declarations.push(endpoint);
69            } else {
70                endpoints.push(endpoint);
71            }
72        }
73    }
74    append_declarations(&declarations, &mut result);
75    append_endpoints(&declarations, &endpoints, &mut result);
76    finish(&mut result);
77    result
78}
79
80fn append_declarations(declarations: &[EventEndpoint], result: &mut EventGraphFacts) {
81    for declaration in declarations {
82        let channel = channel_node(&declaration.observation);
83        let artifact = artifact_node(&declaration.repo_id, &declaration.source_path);
84        result.edges.push(edge(
85            &artifact.id,
86            &channel.id,
87            match declaration.observation.role {
88                EventRole::Publisher => EdgeKind::Publishes,
89                EventRole::Subscriber => EdgeKind::Subscribes,
90                EventRole::Declaration => EdgeKind::Contains,
91            },
92            vec![declaration.evidence.id.clone()],
93        ));
94        append_schema(declaration, &channel, result);
95        append_dead_letter(declaration, &channel, result);
96        result.nodes.push(channel);
97        result.evidence.push(declaration.evidence.clone());
98    }
99}
100
101fn append_endpoints(
102    declarations: &[EventEndpoint],
103    endpoints: &[EventEndpoint],
104    result: &mut EventGraphFacts,
105) {
106    for endpoint in endpoints {
107        let channel = resolve_channel(declarations, endpoint).map_or_else(
108            || channel_node(&endpoint.observation),
109            |declaration| channel_node(&declaration.observation),
110        );
111        let source = source_node(endpoint);
112        match endpoint.observation.role {
113            EventRole::Publisher => result.edges.push(edge(
114                &source.id,
115                &channel.id,
116                EdgeKind::Publishes,
117                vec![endpoint.evidence.id.clone()],
118            )),
119            EventRole::Subscriber => {
120                result.edges.push(edge(
121                    &source.id,
122                    &channel.id,
123                    EdgeKind::Subscribes,
124                    vec![endpoint.evidence.id.clone()],
125                ));
126                result.edges.push(edge(
127                    &channel.id,
128                    &source.id,
129                    EdgeKind::DeliversTo,
130                    vec![endpoint.evidence.id.clone()],
131                ));
132            }
133            EventRole::Declaration => {}
134        }
135        result.nodes.extend([source, channel]);
136        result.evidence.push(endpoint.evidence.clone());
137    }
138}
139
140fn resolve_channel<'a>(
141    declarations: &'a [EventEndpoint],
142    endpoint: &EventEndpoint,
143) -> Option<&'a EventEndpoint> {
144    let channel = endpoint.observation.channel.as_deref()?;
145    let candidates = declarations
146        .iter()
147        .filter(|declaration| {
148            declaration.observation.channel.as_deref() == Some(channel)
149                && (endpoint.observation.broker == EventBroker::Generic
150                    || declaration.observation.broker == endpoint.observation.broker)
151                && endpoint
152                    .observation
153                    .namespace
154                    .as_ref()
155                    .is_none_or(|namespace| {
156                        declaration.observation.namespace.as_ref() == Some(namespace)
157                    })
158        })
159        .collect::<Vec<_>>();
160    let first = candidates.first().copied()?;
161    let identity = channel_node(&first.observation).id;
162    candidates
163        .iter()
164        .all(|candidate| channel_node(&candidate.observation).id == identity)
165        .then_some(first)
166}
167
168fn append_schema(declaration: &EventEndpoint, channel: &Node, result: &mut EventGraphFacts) {
169    let Some(schema) = &declaration.observation.schema else {
170        return;
171    };
172    let Some(name) = schema
173        .name
174        .as_deref()
175        .or(declaration.observation.event_type.as_deref())
176    else {
177        return;
178    };
179    let stable_key = format!(
180        "event-schema:{:?}:{name}:{}",
181        declaration.observation.broker,
182        schema.version.as_deref().unwrap_or("")
183    );
184    let node = Node {
185        id: NodeId::new(stable_id("node", &stable_key)),
186        kind: NodeKind::EventSchema,
187        repo_id: Some(declaration.repo_id.clone()),
188        stable_key,
189        label: name.to_owned(),
190    };
191    result.edges.push(edge(
192        &channel.id,
193        &node.id,
194        EdgeKind::Contains,
195        vec![declaration.evidence.id.clone()],
196    ));
197    result.nodes.push(node);
198}
199
200fn append_dead_letter(declaration: &EventEndpoint, channel: &Node, result: &mut EventGraphFacts) {
201    let Some(dead_letter) = &declaration.observation.dead_letter_channel else {
202        return;
203    };
204    let mut observation = declaration.observation.clone();
205    observation.channel = Some(dead_letter.clone());
206    let target = channel_node(&observation);
207    result.edges.push(edge(
208        &channel.id,
209        &target.id,
210        EdgeKind::CallsRemote,
211        vec![declaration.evidence.id.clone()],
212    ));
213    result.nodes.push(target);
214}
215
216fn source_node(endpoint: &EventEndpoint) -> Node {
217    let channel = endpoint.observation.channel.as_deref().unwrap_or("dynamic");
218    let stable_key = format!(
219        "event-source:{}:{}:{:?}:{channel}:{:?}",
220        endpoint.repo_id.as_str(),
221        endpoint.source_path,
222        endpoint.observation.role,
223        endpoint.observation.broker
224    );
225    Node {
226        id: NodeId::new(stable_id("node", &stable_key)),
227        kind: NodeKind::SymbolRef,
228        repo_id: Some(endpoint.repo_id.clone()),
229        stable_key,
230        label: format!("{} {channel}", endpoint.source_path),
231    }
232}
233
234fn channel_node(observation: &EventObservation) -> Node {
235    let channel = observation.channel.as_deref().unwrap_or("dynamic");
236    let stable_key = format!(
237        "event:{:?}:{}:{channel}",
238        observation.broker,
239        observation.namespace.as_deref().unwrap_or("")
240    );
241    Node {
242        id: NodeId::new(stable_id("node", &stable_key)),
243        kind: NodeKind::EventChannel,
244        repo_id: None,
245        stable_key,
246        label: channel.to_owned(),
247    }
248}
249
250fn artifact_node(repo_id: &RepoId, source_path: &str) -> Node {
251    let stable_key = format!("event-artifact:{}:{source_path}", repo_id.as_str());
252    Node {
253        id: NodeId::new(stable_id("node", &stable_key)),
254        kind: NodeKind::Artifact,
255        repo_id: Some(repo_id.clone()),
256        stable_key,
257        label: source_path.to_owned(),
258    }
259}
260
261fn repository_node(repo_id: &RepoId) -> Node {
262    let stable_key = format!("repository:{}", repo_id.as_str());
263    Node {
264        id: NodeId::new(stable_id("node", &stable_key)),
265        kind: NodeKind::Repository,
266        repo_id: Some(repo_id.clone()),
267        stable_key,
268        label: repo_id.as_str().to_owned(),
269    }
270}
271
272fn event_evidence(
273    repo_id: &RepoId,
274    source_path: &str,
275    content_hash: &str,
276    observation: Option<&EventObservation>,
277    note: &str,
278) -> Evidence {
279    let line = observation
280        .and_then(|observation| observation.evidence.first())
281        .map_or(1, |evidence| evidence.line);
282    let key = format!(
283        "{}:{source_path}:{line}:{note}:{content_hash}",
284        repo_id.as_str()
285    );
286    Evidence {
287        id: EvidenceId::new(stable_id("evidence", &key)),
288        repo_id: Some(repo_id.clone()),
289        file_path: Some(source_path.to_owned()),
290        start_line: Some(line),
291        end_line: Some(line),
292        extractor: "code-system-graph.events".to_owned(),
293        extractor_version: "1.0.0".to_owned(),
294        provenance: Provenance::Extracted,
295        confidence: observation.map_or(1.0, |observation| observation.confidence),
296        observed_at_commit: None,
297        content_hash: Some(content_hash.to_owned()),
298        note: Some(note.to_owned()),
299    }
300}
301
302fn edge(source: &NodeId, target: &NodeId, kind: EdgeKind, evidence: Vec<EvidenceId>) -> Edge {
303    let key = format!("{}:{kind:?}:{}", source.as_str(), target.as_str());
304    Edge {
305        id: EdgeId::new(stable_id("edge", &key)),
306        source: source.clone(),
307        target: target.clone(),
308        kind,
309        confidence: 1.0,
310        status: EpistemicStatus::Confirmed,
311        evidence,
312    }
313}
314
315fn finish(result: &mut EventGraphFacts) {
316    result.nodes.sort_by(|left, right| left.id.cmp(&right.id));
317    result.nodes.dedup_by(|left, right| left.id == right.id);
318    result.edges.sort_by(|left, right| left.id.cmp(&right.id));
319    result.edges.dedup_by(|left, right| left.id == right.id);
320    result
321        .evidence
322        .sort_by(|left, right| left.id.cmp(&right.id));
323    result.evidence.dedup_by(|left, right| left.id == right.id);
324}
325
326#[cfg(test)]
327mod tests {
328    use code_system_graph_model::{EdgeKind, RepoId};
329
330    use super::event_documents_to_graph;
331    use crate::{SourceLanguage, extract_asyncapi, parse_event_source};
332
333    #[test]
334    fn event_graph_should_trace_publisher_through_channel_to_subscriber() {
335        let declaration = extract_asyncapi(
336            "asyncapi.yaml",
337            r"
338asyncapi: 2.6.0
339info: { title: events, version: 1.0.0 }
340servers:
341  kafka: { url: kafka:9092, protocol: kafka }
342channels:
343  orders.created:
344    publish: { message: { name: OrderCreated } }
345",
346        );
347        let publisher = parse_event_source(
348            SourceLanguage::Rust,
349            r#"kafka_producer.send("orders.created", payload);"#,
350        );
351        let subscriber = parse_event_source(
352            SourceLanguage::Python,
353            r#"kafka_consumer.subscribe("orders.created")"#,
354        );
355        let api = RepoId::new("repo:api");
356        let worker = RepoId::new("repo:worker");
357        let facts = declaration.as_ref().ok().map(|declaration| {
358            event_documents_to_graph(&[
359                (&api, "asyncapi.yaml", "contract", declaration),
360                (&api, "src/lib.rs", "publisher", &publisher),
361                (&worker, "worker.py", "subscriber", &subscriber),
362            ])
363        });
364
365        assert!(matches!(
366            facts,
367            Some(facts)
368                if [EdgeKind::Publishes, EdgeKind::Subscribes, EdgeKind::DeliversTo]
369                    .into_iter()
370                    .all(|kind| facts.edges.iter().any(|edge| edge.kind == kind))
371        ));
372    }
373}
code_system_graph_core/event_graph.rs

code_system_graph_core/
event_graph.rs
