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::RecordNodeCard(card) => {
81                super::node_card::project(tx, &card)?;
82            }
83            ProjectionMutation::EnsureNode(node) => {
84                ensure_node(tx, node)?;
85            }
86            ProjectionMutation::UpsertNode(node) => {
87                write_node(tx, node)?;
88            }
89            ProjectionMutation::UpdateNodeStatus { node_id, status } => {
90                update_node_status(tx, &node_id, status)?;
91            }
92            ProjectionMutation::UpsertNodeRelation(relation) => {
93                let NodeRelationProjection {
94                    source_node_id,
95                    target_node_id,
96                    relation_type,
97                    explanation,
98                } = *relation;
99                ensure_node(tx, placeholder_node(&source_node_id))?;
100                ensure_node(tx, placeholder_node(&target_node_id))?;
101                let bytes = encode_explanation(&explanation)?;
102                tx.insert(
103                    Table::Relations,
104                    Key::Str3(&source_node_id, &target_node_id, &relation_type),
105                    &bytes,
106                )?;
107                tx.insert(
108                    Table::RelationsByTarget,
109                    Key::Str3(&target_node_id, &source_node_id, &relation_type),
110                    &[],
111                )?;
112            }
113            ProjectionMutation::RemoveNodeRelation {
114                source_node_id,
115                target_node_id,
116                relation_type,
117            } => {
118                tx.remove(
119                    Table::Relations,
120                    Key::Str3(&source_node_id, &target_node_id, &relation_type),
121                )?;
122                tx.remove(
123                    Table::RelationsByTarget,
124                    Key::Str3(&target_node_id, &source_node_id, &relation_type),
125                )?;
126            }
127            ProjectionMutation::UpsertNodeDetail(detail) => {
128                let node_id = detail.node_id.clone();
129                let bytes = encode("node detail", &DetailRecord::from(detail.clone()))?;
130                // The header describes the bytes this transaction stores, so
131                // it is built here, from them, and never recomputed later.
132                let header = DetailHeaderRecord::describe(&detail, &bytes);
133                tx.insert(Table::Details, Key::Str(&node_id), &bytes)?;
134                detail_header::write(tx, &header)?;
135            }
136        }
137        applied += 1;
138    }
139
140    Ok(applied)
141}
142
143impl ProjectionWriter for EmbeddedKernelStore {
144    async fn apply_mutations(&self, mutations: Vec<ProjectionMutation>) -> Result<(), PortError> {
145        if mutations.is_empty() {
146            return Ok(());
147        }
148        self.run(move |store| {
149            let mut tx = store.begin_write()?;
150            apply_mutations_in_transaction(tx.as_mut(), mutations)?;
151            tx.commit()
152        })
153        .await
154    }
155}
kmp_adapter_embedded/adapter/projection_write.rs

kmp_adapter_embedded/adapter/
projection_write.rs
