kmp_application/queries/
get_node_details.rs

1use kmp_domain::{GraphNeighborhoodReader, NodeDetailReader, PortError};
2
3use super::get_node_detail::{map_node, map_node_detail, trim_to_option};
4use super::{GetNodeDetailResult, QueryApplicationService};
5use crate::ApplicationError;
6
7impl<G, D, S> QueryApplicationService<G, D, S>
8where
9    G: GraphNeighborhoodReader + Send + Sync,
10    D: NodeDetailReader + Send + Sync,
11{
12    /// Materialize already selected refs in order, retaining absent nodes and
13    /// absent bodies as different states. Use this service's operation snapshot
14    /// when graph and bodies must describe the same committed state.
15    pub async fn get_node_details(
16        &self,
17        node_ids: Vec<String>,
18    ) -> Result<Vec<Option<GetNodeDetailResult>>, ApplicationError> {
19        let ids = node_ids
20            .iter()
21            .map(|id| {
22                trim_to_option(id)
23                    .ok_or_else(|| ApplicationError::Validation("node_id cannot be empty".into()))
24            })
25            .collect::<Result<Vec<_>, _>>()?;
26        if ids.is_empty() {
27            return Ok(Vec::new());
28        }
29        let nodes = self.graph_reader.load_nodes_batch(ids.clone()).await?;
30        if nodes.len() != ids.len()
31            || nodes
32                .iter()
33                .zip(&ids)
34                .any(|(node, id)| node.as_ref().is_some_and(|node| node.node_id != *id))
35        {
36            return Err(invalid_batch("graph"));
37        }
38        // A dangling link is not a source. Do not read orphaned bodies whose
39        // graph node is absent; the single-node operation skips them too.
40        let present: Vec<_> = nodes
41            .iter()
42            .flatten()
43            .map(|node| node.node_id.clone())
44            .collect();
45        let details = if present.is_empty() {
46            Vec::new()
47        } else {
48            self.detail_reader
49                .load_node_details_batch(present.clone())
50                .await?
51        };
52        if details.len() != present.len()
53            || details
54                .iter()
55                .zip(&present)
56                .any(|(detail, id)| detail.as_ref().is_some_and(|detail| detail.node_id != *id))
57        {
58            return Err(invalid_batch("detail"));
59        }
60        let mut details = details.into_iter();
61        Ok(nodes
62            .into_iter()
63            .map(|node| {
64                node.map(|node| GetNodeDetailResult {
65                    node: map_node(&node),
66                    detail: details
67                        .next()
68                        .expect("validated batch length")
69                        .map(map_node_detail),
70                })
71            })
72            .collect())
73    }
74}
75
76fn invalid_batch(port: &str) -> ApplicationError {
77    PortError::InvalidState(format!(
78        "{port} batch does not preserve requested node slots"
79    ))
80    .into()
81}
kmp_application/queries/get_node_details.rs

kmp_application/queries/
get_node_details.rs
