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

kmp_adapter_embedded/adapter/
projection_write.rs
