kmp_adapter_embedded/adapter/
projection_write.rs

1use std::collections::BTreeMap;
2
3use kmp_domain::{
4    NodeProjection, NodeRelationProjection, PortError, ProjectionMutation, ProjectionWriter,
5};
6
7use super::engine::{Key, Table, WriteTx};
8use super::serdes::{DetailRecord, NodeRecord, decode, encode, encode_explanation};
9use super::store::EmbeddedKernelStore;
10
11pub(crate) const MEMORY_ANCHOR_KIND: &str = "memory_anchor";
12
13/// Mirrors the Neo4j relation-upsert `MERGE ... ON CREATE` placeholder so the
14/// conformance suite observes identical semantics across backends.
15fn placeholder_node(node_id: &str) -> NodeProjection {
16    NodeProjection {
17        node_id: node_id.to_string(),
18        node_kind: "placeholder".to_string(),
19        title: "[unmaterialized node]".to_string(),
20        summary: "Referenced by relation before node materialization".to_string(),
21        status: "UNMATERIALIZED".to_string(),
22        labels: vec!["placeholder".to_string()],
23        properties: BTreeMap::from([
24            ("placeholder".to_string(), "true".to_string()),
25            (
26                "placeholder_reason".to_string(),
27                "relation_materialized_before_node".to_string(),
28            ),
29            (
30                "placeholder_created_by_subject".to_string(),
31                "graph.relation.materialized".to_string(),
32            ),
33        ]),
34        provenance: None,
35    }
36}
37
38fn write_node(tx: &mut dyn WriteTx, node: NodeProjection) -> Result<(), PortError> {
39    let node_id = node.node_id.clone();
40    let is_anchor = node.node_kind == MEMORY_ANCHOR_KIND;
41    let bytes = encode("graph node", &NodeRecord::from(node))?;
42    tx.insert(Table::Nodes, Key::Str(&node_id), &bytes)?;
43    if is_anchor {
44        tx.insert(Table::Anchors, Key::Str(&node_id), &[])
45    } else {
46        tx.remove(Table::Anchors, Key::Str(&node_id))
47    }
48}
49
50fn ensure_node(tx: &mut dyn WriteTx, node: NodeProjection) -> Result<(), PortError> {
51    if tx.get(Table::Nodes, Key::Str(&node.node_id))?.is_some() {
52        return Ok(());
53    }
54    write_node(tx, node)
55}
56
57fn update_node_status(
58    tx: &mut dyn WriteTx,
59    node_id: &str,
60    status: String,
61) -> Result<(), PortError> {
62    let bytes = tx.get(Table::Nodes, Key::Str(node_id))?.ok_or_else(|| {
63        PortError::InvalidState(format!("cannot update missing node `{node_id}`"))
64    })?;
65    let mut node = decode::<NodeRecord>("graph node", &bytes)?.into_projection()?;
66    node.status = status;
67    write_node(tx, node)
68}
69
70/// Applies a mutation batch inside one transaction. Shared by the live write
71/// path and the replay tool so both apply byte-identical projections.
72pub(crate) fn apply_mutations_in_transaction(
73    tx: &mut dyn WriteTx,
74    mutations: Vec<ProjectionMutation>,
75) -> Result<u64, PortError> {
76    let mut applied = 0u64;
77    for mutation in mutations {
78        match mutation {
79            ProjectionMutation::EnsureNode(node) => {
80                ensure_node(tx, node)?;
81            }
82            ProjectionMutation::UpsertNode(node) => {
83                write_node(tx, node)?;
84            }
85            ProjectionMutation::UpdateNodeStatus { node_id, status } => {
86                update_node_status(tx, &node_id, status)?;
87            }
88            ProjectionMutation::UpsertNodeRelation(relation) => {
89                let NodeRelationProjection {
90                    source_node_id,
91                    target_node_id,
92                    relation_type,
93                    explanation,
94                } = *relation;
95                ensure_node(tx, placeholder_node(&source_node_id))?;
96                ensure_node(tx, placeholder_node(&target_node_id))?;
97                let bytes = encode_explanation(&explanation)?;
98                tx.insert(
99                    Table::Relations,
100                    Key::Str3(&source_node_id, &target_node_id, &relation_type),
101                    &bytes,
102                )?;
103                tx.insert(
104                    Table::RelationsByTarget,
105                    Key::Str3(&target_node_id, &source_node_id, &relation_type),
106                    &[],
107                )?;
108            }
109            ProjectionMutation::UpsertNodeDetail(detail) => {
110                let node_id = detail.node_id.clone();
111                let bytes = encode("node detail", &DetailRecord::from(detail))?;
112                tx.insert(Table::Details, Key::Str(&node_id), &bytes)?;
113            }
114        }
115        applied += 1;
116    }
117
118    Ok(applied)
119}
120
121impl ProjectionWriter for EmbeddedKernelStore {
122    async fn apply_mutations(&self, mutations: Vec<ProjectionMutation>) -> Result<(), PortError> {
123        if mutations.is_empty() {
124            return Ok(());
125        }
126        self.run(move |store| {
127            let mut tx = store.begin_write()?;
128            apply_mutations_in_transaction(tx.as_mut(), mutations)?;
129            tx.commit()
130        })
131        .await
132    }
133}
kmp_adapter_embedded/adapter/projection_write.rs

kmp_adapter_embedded/adapter/
projection_write.rs
