kmp_adapter_embedded/adapter/
projection_write.rs

1use std::collections::BTreeMap;
2
3use kmp_domain::{
4    NodeProjection, NodeRelationProjection, PortError, ProjectionMutation, ProjectionWriter,
5};
6use redb::{ReadableTable, WriteTransaction};
7
8use super::serdes::{DetailRecord, NodeRecord, encode, encode_explanation};
9use super::store::{
10    ANCHORS, DETAILS, EmbeddedKernelStore, NODES, RELATIONS, RELATIONS_BY_TARGET, commit_error,
11    storage_error, table_error,
12};
13
14pub(crate) const MEMORY_ANCHOR_KIND: &str = "memory_anchor";
15
16type NodeTable<'txn> = redb::Table<'txn, &'static str, &'static [u8]>;
17type AnchorTable<'txn> = redb::Table<'txn, &'static str, ()>;
18
19/// Mirrors the Neo4j relation-upsert `MERGE ... ON CREATE` placeholder so the
20/// conformance suite observes identical semantics across backends.
21fn placeholder_node(node_id: &str) -> NodeProjection {
22    NodeProjection {
23        node_id: node_id.to_string(),
24        node_kind: "placeholder".to_string(),
25        title: "[unmaterialized node]".to_string(),
26        summary: "Referenced by relation before node materialization".to_string(),
27        status: "UNMATERIALIZED".to_string(),
28        labels: vec!["placeholder".to_string()],
29        properties: BTreeMap::from([
30            ("placeholder".to_string(), "true".to_string()),
31            (
32                "placeholder_reason".to_string(),
33                "relation_materialized_before_node".to_string(),
34            ),
35            (
36                "placeholder_created_by_subject".to_string(),
37                "graph.relation.materialized".to_string(),
38            ),
39        ]),
40        provenance: None,
41    }
42}
43
44fn write_node(
45    nodes: &mut NodeTable<'_>,
46    anchors: &mut AnchorTable<'_>,
47    node: NodeProjection,
48) -> Result<(), PortError> {
49    let node_id = node.node_id.clone();
50    let is_anchor = node.node_kind == MEMORY_ANCHOR_KIND;
51    let bytes = encode("graph node", &NodeRecord::from(node))?;
52    nodes
53        .insert(node_id.as_str(), bytes.as_slice())
54        .map_err(storage_error)?;
55    if is_anchor {
56        anchors
57            .insert(node_id.as_str(), ())
58            .map_err(storage_error)?;
59    } else {
60        anchors.remove(node_id.as_str()).map_err(storage_error)?;
61    }
62    Ok(())
63}
64
65fn ensure_node(
66    nodes: &mut NodeTable<'_>,
67    anchors: &mut AnchorTable<'_>,
68    node: NodeProjection,
69) -> Result<(), PortError> {
70    if nodes
71        .get(node.node_id.as_str())
72        .map_err(storage_error)?
73        .is_some()
74    {
75        return Ok(());
76    }
77    write_node(nodes, anchors, node)
78}
79
80/// Applies a mutation batch inside one transaction. Shared by the live write
81/// path and the replay tool so both apply byte-identical projections.
82pub(crate) fn apply_mutations_in_transaction(
83    tx: &WriteTransaction,
84    mutations: Vec<ProjectionMutation>,
85) -> Result<u64, PortError> {
86    let mut nodes = tx.open_table(NODES).map_err(table_error)?;
87    let mut anchors = tx.open_table(ANCHORS).map_err(table_error)?;
88    let mut relations = tx.open_table(RELATIONS).map_err(table_error)?;
89    let mut relations_by_target = tx.open_table(RELATIONS_BY_TARGET).map_err(table_error)?;
90    let mut details = tx.open_table(DETAILS).map_err(table_error)?;
91
92    let mut applied = 0u64;
93    for mutation in mutations {
94        match mutation {
95            ProjectionMutation::EnsureNode(node) => {
96                ensure_node(&mut nodes, &mut anchors, node)?;
97            }
98            ProjectionMutation::UpsertNode(node) => {
99                write_node(&mut nodes, &mut anchors, node)?;
100            }
101            ProjectionMutation::UpsertNodeRelation(relation) => {
102                let NodeRelationProjection {
103                    source_node_id,
104                    target_node_id,
105                    relation_type,
106                    explanation,
107                } = *relation;
108                ensure_node(&mut nodes, &mut anchors, placeholder_node(&source_node_id))?;
109                ensure_node(&mut nodes, &mut anchors, placeholder_node(&target_node_id))?;
110                let bytes = encode_explanation(&explanation)?;
111                relations
112                    .insert(
113                        (
114                            source_node_id.as_str(),
115                            target_node_id.as_str(),
116                            relation_type.as_str(),
117                        ),
118                        bytes.as_slice(),
119                    )
120                    .map_err(storage_error)?;
121                relations_by_target
122                    .insert(
123                        (
124                            target_node_id.as_str(),
125                            source_node_id.as_str(),
126                            relation_type.as_str(),
127                        ),
128                        (),
129                    )
130                    .map_err(storage_error)?;
131            }
132            ProjectionMutation::UpsertNodeDetail(detail) => {
133                let node_id = detail.node_id.clone();
134                let bytes = encode("node detail", &DetailRecord::from(detail))?;
135                details
136                    .insert(node_id.as_str(), bytes.as_slice())
137                    .map_err(storage_error)?;
138            }
139        }
140        applied += 1;
141    }
142
143    Ok(applied)
144}
145
146impl ProjectionWriter for EmbeddedKernelStore {
147    async fn apply_mutations(&self, mutations: Vec<ProjectionMutation>) -> Result<(), PortError> {
148        if mutations.is_empty() {
149            return Ok(());
150        }
151        self.run(move |store| {
152            let tx = store.begin_write()?;
153            apply_mutations_in_transaction(&tx, mutations)?;
154            tx.commit().map_err(commit_error)
155        })
156        .await
157    }
158}
kmp_adapter_embedded/adapter/projection_write.rs

kmp_adapter_embedded/adapter/
projection_write.rs
