kmp_application/queries/
get_node_details.rs1use 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 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 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}