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, 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
57/// Applies a mutation batch inside one transaction. Shared by the live write
58/// path and the replay tool so both apply byte-identical projections.
59pub(crate) fn apply_mutations_in_transaction(
60    tx: &mut dyn WriteTx,
61    mutations: Vec<ProjectionMutation>,
62) -> Result<u64, PortError> {
63    let mut applied = 0u64;
64    for mutation in mutations {
65        match mutation {
66            ProjectionMutation::EnsureNode(node) => {
67                ensure_node(tx, node)?;
68            }
69            ProjectionMutation::UpsertNode(node) => {
70                write_node(tx, node)?;
71            }
72            ProjectionMutation::UpsertNodeRelation(relation) => {
73                let NodeRelationProjection {
74                    source_node_id,
75                    target_node_id,
76                    relation_type,
77                    explanation,
78                } = *relation;
79                ensure_node(tx, placeholder_node(&source_node_id))?;
80                ensure_node(tx, placeholder_node(&target_node_id))?;
81                let bytes = encode_explanation(&explanation)?;
82                tx.insert(
83                    Table::Relations,
84                    Key::Str3(&source_node_id, &target_node_id, &relation_type),
85                    &bytes,
86                )?;
87                tx.insert(
88                    Table::RelationsByTarget,
89                    Key::Str3(&target_node_id, &source_node_id, &relation_type),
90                    &[],
91                )?;
92            }
93            ProjectionMutation::UpsertNodeDetail(detail) => {
94                let node_id = detail.node_id.clone();
95                let bytes = encode("node detail", &DetailRecord::from(detail))?;
96                tx.insert(Table::Details, Key::Str(&node_id), &bytes)?;
97            }
98        }
99        applied += 1;
100    }
101
102    Ok(applied)
103}
104
105impl ProjectionWriter for EmbeddedKernelStore {
106    async fn apply_mutations(&self, mutations: Vec<ProjectionMutation>) -> Result<(), PortError> {
107        if mutations.is_empty() {
108            return Ok(());
109        }
110        self.run(move |store| {
111            let mut tx = store.begin_write()?;
112            apply_mutations_in_transaction(tx.as_mut(), mutations)?;
113            tx.commit()
114        })
115        .await
116    }
117}
kmp_adapter_embedded/adapter/projection_write.rs

kmp_adapter_embedded/adapter/
projection_write.rs
