streaming_telemetry/src/handlers/chassis.rs - streaming-telemetry-server - Git at Google

 // Copyright 2025 Google LLC
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
 //      http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and
 // limitations under the License.

 //! This module handles chassis-related operations, including chassis collection, individual chassis,
 //! sensors collection, and individual sensors. It also includes functionality for event subscription
 //! and processing.

 // This is for using Json name directly
 #![allow(non_snake_case)]

 use axum::body::to_bytes;
 use axum::{
     extract::{Json, Path, Query, State},
     http::StatusCode,
     response::IntoResponse,
 };
 use dashmap::DashMap;
 use futures::future::join_all;
 use serde_json::json;
 use serde_json::Value;
 use std::convert::Infallible;
 use std::sync::Arc;

 use axum::{extract::OriginalUri, http::Uri};

 use redfish_codegen::models::odata_v4;
 use redfish_codegen::models::{
     chassis::v1_23_0::Actions as ChassisActions, chassis::v1_23_0::Chassis as ChassisModel,
     chassis::v1_23_0::ChassisType, chassis::v1_23_0::Reset as ChassisReset,
     chassis_collection::ChassisCollection as ChassisCollectionModel, resource,
     sensor::v1_7_0::Sensor as SensorModel,
     sensor_collection::SensorCollection as SensorCollectionModel,
 };

 use crate::dbus_client::{get_objects_with_interface, get_one_sensor, get_sensors_under_a_chassis};

 use crate::app_state::AppState;
 use crate::composite_query::{RequestResponse, SubRequest};

 /// Represents the state of the chassis, including a cache for chassis data.
 #[derive(Debug, Default)]
 pub struct ChassisState {
     pub cache: DashMap<String, serde_json::Value>,
 }

 /// Represents the query parameters for expanding resources.
 #[derive(serde::Deserialize, Debug, Default)]
 struct ExpandQuery {
     #[serde(rename = "$expand")]
     expand: Option<String>,
 }

 /// Retrieves all chassis from the system.
 ///
 /// # Returns
 ///
 /// A Result containing a vector of chassis names or an error.
 async fn get_all_chassis() -> Result<Vec<String>, Box<dyn std::error::Error>> {
     let result = get_objects_with_interface(
         Vec::new(),
         "/xyz/openbmc_project",
         "xyz.openbmc_project.Inventory.Item.Board",
     )
     .await?;
     let mut chassis: Vec<String> = Vec::new();
     for (_service, objects) in result {
         for object_path in objects {
             let path_segments: Vec<&str> = object_path.split('/').collect();
             if let Some(last_segment) = path_segments.last() {
                 chassis.push(last_segment.to_string());
             }
         }
     }
     Ok(chassis)
 }

 /// Handles requests for the chassis collection.
 ///
 /// # Arguments
 ///
 /// * `state` - The application state.
 /// * `uri` - The original URI of the request.
 /// * `params` - Query parameters for expansion.
 async fn chassis_collection(
     State(state): State<Arc<AppState>>,
     uri: OriginalUri,
     Query(params): Query<ExpandQuery>,
 ) -> impl IntoResponse {
     let params_str = params.expand.clone().unwrap_or_default();
     let cache_key = format!("{}?{}", uri.path(), params_str);
     if let Some(cached_response) = state.chassis_state.cache.get(&cache_key) {
         return (StatusCode::OK, Json(cached_response.value().clone()));
     }

     let chassis_collection = get_all_chassis().await.unwrap_or_default();
     let members = chassis_collection
         .into_iter()
         .map(|chassis_id| format!("{}/{}", uri.path(), chassis_id))
         .map(|chassis_uri| odata_v4::IdRef {
             odata_id: Some(odata_v4::Id(chassis_uri)),
         })
         .collect::<Vec<_>>();
     let response = ChassisCollectionModel {
         odata_id: odata_v4::Id(uri.path().to_string()),
         members_odata_count: odata_v4::Count(members.len() as i64),
         members,
         name: resource::Name("Chassis Collection".to_string()),
         ..Default::default()
     };

     let mut response_json = serde_json::to_value(response).unwrap_or_default();
     let expand = params.expand.as_deref() == Some(".($levels=1)");
     if expand {
         let member_uris: Vec<String> = response_json["Members"]
             .as_array()
             .unwrap_or(&Vec::new())
             .iter()
             .filter_map(|m| m["@odata.id"].as_str().map(ToString::to_string))
             .collect();

         let tasks = member_uris
             .into_iter()
             .map(|odata_id| {
                 let state = state.clone();
                 tokio::spawn(async move {
                     let uri: Uri = odata_id.parse().unwrap_or(Uri::default());
                     let chassis_id = odata_id.split('/').last().unwrap_or("");
                     let response = chassis(
                         State(state),
                         OriginalUri(uri),
                         Path(chassis_id.to_string()),
                         Query::default(),
                     )
                     .await
                     .into_response();
                     let body_bytes = to_bytes(response.into_body(), usize::MAX)
                         .await
                         .unwrap_or_default();
                     serde_json::from_slice::<Value>(&body_bytes).unwrap_or_default()
                 })
             })
             .collect::<Vec<_>>();
         let results = futures::future::join_all(tasks).await;

         if let Some(members) = response_json
             .get_mut("Members")
             .and_then(|m| m.as_array_mut())
         {
             for (member, result) in members.iter_mut().zip(results) {
                 match result {
                     Ok(expanded_sensor) => *member = expanded_sensor,
                     Err(_e) => {
                         *member = json!({"error": "Failed to expand sensor"});
                     }
                 }
             }
         }
     }

     // TODO: create async task to monitoring the changes related to this cache entry
     state
         .chassis_state
         .cache
         .insert(cache_key, response_json.clone());
     (StatusCode::OK, Json(response_json))
 }

 /// Handles requests for individual chassis.
 ///
 /// # Arguments
 ///
 /// * `state` - The application state.
 /// * `uri` - The original URI of the request.
 /// * `chassis_id` - The ID of the chassis.
 /// * `params` - Query parameters for expansion.
 async fn chassis(
     State(state): State<Arc<AppState>>,
     uri: OriginalUri,
     Path(chassis_id): Path<String>,
     Query(params): Query<ExpandQuery>,
 ) -> impl IntoResponse {
     let params_str = params.expand.clone().unwrap_or_default();
     let cache_key = format!("{}?{}", uri.path(), params_str);
     if let Some(cached_response) = state.chassis_state.cache.get(&cache_key) {
         return (StatusCode::OK, Json(cached_response.value().clone()));
     }

     let response = ChassisModel {
         odata_id: odata_v4::Id(uri.path().to_string()),
         actions: Some(ChassisActions {
             chassis_reset: Some(ChassisReset {
                 target: Some(format!("{}/{}", uri.path(), "Actions/Chassis.Reset")),
                 ..Default::default()
             }),
             ..Default::default()
         }),
         sensors: Some(odata_v4::IdRef {
             odata_id: Some(odata_v4::Id(format!("{}/{}", uri.path(), "Sensors"))),
         }),
         chassis_type: ChassisType::RackMount,
         id: resource::Id(chassis_id.clone()),
         name: resource::Name(format!("Chassis {}", chassis_id)),
         ..Default::default()
     };

     // TODO: create async task to monitoring the changes related to this cache entry
     let response_json = serde_json::to_value(response).unwrap_or_default();
     state
         .chassis_state
         .cache
         .insert(cache_key, response_json.clone());
     (StatusCode::OK, Json(response_json))
 }

 /// Retrieves the object path for all sensors under a chassis.
 ///
 /// # Arguments
 ///
 /// * `sensor_collection_uri` - The URI of the sensor collection.
 ///
 /// # Returns
 ///
 /// A string representing the object path for all sensors.
 fn get_all_sensors_object_path(sensor_collection_uri: &OriginalUri) -> String {
     let path = sensor_collection_uri.path().to_string();
     let parts: Vec<&str> = path.split('/').collect();
     if let Some(chassis_index) = parts.iter().position(|&r| r == "Chassis") {
         if let Some(chassis_name) = parts.get(chassis_index + 1) {
             return format!(
                 "/xyz/openbmc_project/inventory/system/board/{}/all_sensors",
                 chassis_name
             );
         }
     }
     String::new()
 }

 /// Handles requests for the sensors collection under a chassis.
 ///
 /// # Arguments
 ///
 /// * `state` - The application state.
 /// * `uri` - The original URI of the request.
 /// * `chassis_id` - The ID of the chassis.
 /// * `params` - Query parameters for expansion.
 async fn sensors_collection(
     State(state): State<Arc<AppState>>,
     uri: OriginalUri,
     Path(chassis_id): Path<String>,
     Query(params): Query<ExpandQuery>,
 ) -> impl IntoResponse {
     let params_str = params.expand.clone().unwrap_or_default();
     let cache_key = format!("{}?{}", uri.path(), params_str);
     if let Some(cached_response) = state.chassis_state.cache.get(&cache_key) {
         return (StatusCode::OK, Json(cached_response.value().clone()));
     }

     let sensors = get_sensors_under_a_chassis(&get_all_sensors_object_path(&uri))
         .await
         .unwrap_or_default();
     let members = sensors
         .into_iter()
         .map(|sensor_id| format!("{}/{}", uri.path(), sensor_id))
         .map(|sensor_uri| odata_v4::IdRef {
             odata_id: Some(odata_v4::Id(sensor_uri)),
         })
         .collect::<Vec<_>>();

     let response = SensorCollectionModel {
         odata_id: odata_v4::Id(uri.path().to_string()),
         members_odata_count: odata_v4::Count(members.len() as i64),
         members,
         name: resource::Name("Sensor Collection".to_string()),
         ..Default::default()
     };

     let mut response_json = serde_json::to_value(response).unwrap_or_default();
     let expand = params.expand.as_deref() == Some(".($levels=1)");
     if expand {
         let member_uris: Vec<String> = response_json["Members"]
             .as_array()
             .unwrap_or(&Vec::new())
             .iter()
             .filter_map(|m| m["@odata.id"].as_str().map(ToString::to_string))
             .collect();

         let tasks = member_uris
             .into_iter()
             .map(|odata_id| {
                 let state = state.clone();
                 let chassis_id = chassis_id.clone();
                 tokio::spawn(async move {
                     let uri: Uri = odata_id.parse().unwrap_or(Uri::default());
                     let sensor_id = odata_id.split('/').last().unwrap_or("");
                     let response = sensor(
                         State(state),
                         OriginalUri(uri),
                         Path((chassis_id.to_string(), sensor_id.to_string())),
                         Query::default(),
                     )
                     .await
                     .into_response();
                     let body_bytes = to_bytes(response.into_body(), usize::MAX)
                         .await
                         .unwrap_or_default();
                     serde_json::from_slice::<Value>(&body_bytes).unwrap_or_default()
                 })
             })
             .collect::<Vec<_>>();
         let results = futures::future::join_all(tasks).await;

         if let Some(members) = response_json
             .get_mut("Members")
             .and_then(|m| m.as_array_mut())
         {
             for (member, result) in members.iter_mut().zip(results) {
                 match result {
                     Ok(expanded_sensor) => *member = expanded_sensor,
                     Err(_e) => {
                         *member = json!({"error": "Failed to expand sensoris"});
                     }
                 }
             }
         }
     }

     // TODO: create async task to monitoring the changes related to this cache entry
     state
         .chassis_state
         .cache
         .insert(cache_key, response_json.clone());
     (StatusCode::OK, Json(response_json))
 }

 /// Formats a sensor name by removing underscores and trimming.
 ///
 /// # Arguments
 ///
 /// * `sensor_name` - The original sensor name.
 ///
 /// # Returns
 ///
 /// A formatted string representing the sensor name.
 fn format_sensor_name(sensor_name: &str) -> String {
     let trimmed_name = sensor_name
         .find('_')
         .map(|index| &sensor_name[index + 1..])
         .unwrap_or(sensor_name);

     trimmed_name.replace('_', " ")
 }

 /// Extracts the chassis ID from a base URL.
 ///
 /// # Arguments
 ///
 /// * `base_url` - The base URL containing the chassis ID.
 ///
 /// # Returns
 ///
 /// An Option containing the chassis ID as a String, if found.
 fn extract_chassis_id(base_url: &str) -> Option<String> {
     let segments: Vec<&str> = base_url.split('/').collect();

     segments
         .iter()
         .position(|&segment| segment == "Chassis")
         .and_then(|index| segments.get(index + 1))
         .map(|&segment| segment.to_string())
 }

 /// Handles a single request in the context of a composite query.
 ///
 /// # Arguments
 ///
 /// * `state` - The application state.
 /// * `sub_request` - The sub-request to handle.
 /// * `context` - The context for storing request results.
 ///
 /// # Returns
 ///
 /// A Result containing the RequestResponse or an Infallible error.
 async fn handle_one_request(
     state: &Arc<AppState>,
     sub_request: &mut SubRequest,
     context: &mut Arc<DashMap<String, Value>>,
 ) -> Result<RequestResponse, Infallible> {
     let (base_url, _) = sub_request
         .url
         .split_at(sub_request.url.find('?').unwrap_or(sub_request.url.len()));
     let segments: Vec<&str> = base_url.split('/').collect();

     match sub_request.method.as_str() {
         "GET" if base_url.ends_with("/Chassis") => {
             if let Ok(parsed_url) = url::Url::parse(&format!("http://localhost{}", sub_request.url))
             {
                 let query: String = parsed_url.query().unwrap_or("").to_string();
                 let expanded_query: ExpandQuery = serde_qs::from_str(&query).unwrap_or_default();
                 let url = sub_request.url.parse().unwrap_or_default();
                 let url = OriginalUri(url);
                 let response = chassis_collection(State(state.clone()), url, Query(expanded_query))
                     .await
                     .into_response();
                 let body_bytes = to_bytes(response.into_body(), usize::MAX)
                     .await
                     .unwrap_or_default();
                 let json = serde_json::from_slice::<Value>(&body_bytes).unwrap_or_default();
                 context.insert(sub_request.referenceId.clone(), json.clone());
                 Ok(RequestResponse::Json(json))
             } else {
                 Ok(RequestResponse::Json(
                     json!({"error": "Invalid URL format"}),
                 ))
             }
         }
         "GET" if segments.len() > 2 && segments[segments.len() - 2] == "Chassis" => {
             let url = sub_request.url.parse().unwrap_or_default();
             let url = OriginalUri(url);
             let chassis_id = extract_chassis_id(base_url).unwrap_or_default();
             let state = state.clone();
             let response = chassis(
                 State(state),
                 url,
                 Path(chassis_id.to_string()),
                 Query::default(),
             )
             .await
             .into_response();
             let body_bytes = to_bytes(response.into_body(), usize::MAX)
                 .await
                 .unwrap_or_default();
             let json = serde_json::from_slice::<Value>(&body_bytes).unwrap_or_default();
             context.insert(sub_request.referenceId.clone(), json.clone());
             Ok(RequestResponse::Json(json))
         }
         "GET" if base_url.ends_with("/Sensors") => {
             if let Ok(parsed_url) = url::Url::parse(&format!("http://localhost{}", sub_request.url))
             {
                 let query: String = parsed_url.query().unwrap_or("").to_string();
                 // Deserialize the query string into ExpandQuery
                 let expanded_query: ExpandQuery = serde_qs::from_str(&query).unwrap_or_default();
                 let url = sub_request.url.parse().unwrap_or_default();
                 let url = OriginalUri(url);
                 let chassis_id = extract_chassis_id(base_url).unwrap_or_default();
                 let response = sensors_collection(
                     State(state.clone()),
                     url,
                     Path(chassis_id),
                     Query(expanded_query),
                 )
                 .await
                 .into_response();
                 let body_bytes = to_bytes(response.into_body(), usize::MAX)
                     .await
                     .unwrap_or_default();
                 let json = serde_json::from_slice::<Value>(&body_bytes).unwrap_or_default();
                 context.insert(sub_request.referenceId.clone(), json.clone());
                 Ok(RequestResponse::Json(json))
             } else {
                 Ok(RequestResponse::Json(
                     json!({"error": "Invalid URL format"}),
                 ))
             }
         }
         "GET" if segments.len() > 2 && segments[segments.len() - 2] == "Sensors" => {
             let url = sub_request.url.parse().unwrap_or_default();
             let url = OriginalUri(url);
             let chassis_id = extract_chassis_id(base_url).unwrap_or_default();
             let sensor_id = base_url.split('/').last().unwrap_or("");
             let state = state.clone();
             let response = sensor(
                 State(state),
                 url,
                 Path((chassis_id.to_string(), sensor_id.to_string())),
                 Query::default(),
             )
             .await
             .into_response();
             let body_bytes = to_bytes(response.into_body(), usize::MAX)
                 .await
                 .unwrap_or_default();
             let json = serde_json::from_slice::<Value>(&body_bytes).unwrap_or_default();
             context.insert(sub_request.referenceId.clone(), json.clone());
             Ok(RequestResponse::Json(json))
         }
         _ => Ok::<RequestResponse, Infallible>(RequestResponse::Json(
             json!({"error": "Unsupported request method or URL"}),
         )),
     }
 }

 /// Retrieves a SensorModel, either from cache or by creating a new one.
 ///
 /// # Arguments
 ///
 /// * `state` - The application state.
 /// * `cache_key` - The key for caching the sensor model.
 /// * `odata_id` - The odata ID of the sensor.
 ///
 /// # Returns
 ///
 /// A Result containing the SensorModel or an Infallible error.
 async fn get_sensor_model(
     state: &Arc<AppState>,
     cache_key: &str,
     odata_id: &str,
 ) -> Result<SensorModel, Infallible> {
     match state.chassis_state.cache.get(cache_key) {
         Some(cached_response) => {
             Ok(serde_json::from_value(cached_response.value().clone()).unwrap_or_default())
         }
         None => {
             let uri: Uri = odata_id.parse().unwrap_or(Uri::default());
             let uri = OriginalUri(uri);
             let sensor_id = uri.path().split('/').last().unwrap_or_default().to_string();
             let mut sensor_model = SensorModel {
                 odata_id: odata_v4::Id(odata_id.to_string()),
                 odata_type: odata_v4::Type("#Sensor.v1_7_0.Sensor".to_string()),
                 id: resource::Id(sensor_id.to_string()),
                 name: resource::Name(format_sensor_name(&sensor_id)),
                 ..Default::default()
             };
             get_one_sensor(&sensor_id, &mut sensor_model)
                 .await
                 .unwrap_or_default();
             Ok(sensor_model)
         }
     }
 }

 /// Handles requests for individual sensors.
 ///
 /// # Arguments
 ///
 /// * `state` - The application state.
 /// * `uri` - The original URI of the request.
 /// * `_chassis_id` - The ID of the chassis (unused).
 /// * `_sensor_id` - The ID of the sensor (unused).
 /// * `_params` - Query parameters for expansion (unused).
 async fn sensor(
     State(state): State<Arc<AppState>>,
     uri: OriginalUri,
     Path((_chassis_id, _sensor_id)): Path<(String, String)>,
     Query(_params): Query<ExpandQuery>,
 ) -> impl IntoResponse {
     let cache_key = uri.path().to_string();
     let sensor_model = get_sensor_model(&state, &cache_key, &cache_key)
         .await
         .unwrap_or_default();
     let response_json = serde_json::to_value(sensor_model).unwrap_or_default();
     state
         .chassis_state
         .cache
         .insert(cache_key, response_json.clone());
     (StatusCode::OK, Json(response_json))
 }

 /// Handles an XPath segment in the context of URL expansion.
 ///
 /// # Arguments
 ///
 /// * `urls` - The vector of URLs to modify.
 /// * `_segment` - The XPath segment (unused).
 /// * `state` - The application state.
 ///
 /// # Returns
 ///
 /// A Result containing a vector of RequestResponse objects or an Infallible error.
 pub async fn handle_xpath_segment(
     urls: &mut Vec<String>,
     _segment: &str,
     state: &Arc<AppState>,
 ) -> Result<Vec<RequestResponse>, Infallible> {
     let context = Arc::new(DashMap::new());
     let futures: Vec<_> = urls
         .iter()
         .map(|url| {
             let state_clone = state.clone();
             let mut context_clone = context.clone();
             async move {
                 let mut request = SubRequest {
                     method: "GET".to_string(),
                     url: url.to_string(),
                     ..Default::default()
                 };

                 let response =
                     handle_one_request(&state_clone, &mut request, &mut context_clone).await;
                 let new_urls = match &response {
                     Ok(RequestResponse::Json(json)) => {
                         // TODO: use segment as query here
                         let query = "$.Members[*]['@odata.id']";
                         jsonpath_lib::select(json, query)
                             .unwrap_or_else(|_| vec![])
                             .iter()
                             .filter_map(|v| v.as_str().map(String::from))
                             .collect::<Vec<String>>()
                     }
                     _ => vec![],
                 };
                 (response, new_urls)
             }
         })
         .collect();

     let results = join_all(futures).await;

     *urls = vec![];
     let mut responses = vec![];
     for (response, new_urls) in results {
         urls.extend(new_urls);
         responses.push(response.unwrap_or_default());
     }

     Ok(responses)
 }

 /// Appends a segment to each URL in the provided vector.
 ///
 /// # Arguments
 ///
 /// * `urls` - The vector of URLs to modify.
 /// * `segment` - The segment to append to each URL.
 pub fn append_to_urls(urls: &mut [String], segment: &str) {
     urls.iter_mut().for_each(|url| {
         if !url.ends_with('/') && !segment.starts_with('/') {
             url.push('/');
         }
         url.push_str(segment);
     });
 }
	// Copyright 2025 Google LLC
	//
	// Licensed under the Apache License, Version 2.0 (the "License");
	// you may not use this file except in compliance with the License.
	// You may obtain a copy of the License at
	//
	// http://www.apache.org/licenses/LICENSE-2.0
	//
	// Unless required by applicable law or agreed to in writing, software
	// distributed under the License is distributed on an "AS IS" BASIS,
	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	// See the License for the specific language governing permissions and
	// limitations under the License.

	//! This module handles chassis-related operations, including chassis collection, individual chassis,
	//! sensors collection, and individual sensors. It also includes functionality for event subscription
	//! and processing.

	// This is for using Json name directly
	#![allow(non_snake_case)]

	use axum::body::to_bytes;
	use axum::{
	extract::{Json, Path, Query, State},
	http::StatusCode,
	response::IntoResponse,
	};
	use dashmap::DashMap;
	use futures::future::join_all;
	use serde_json::json;
	use serde_json::Value;
	use std::convert::Infallible;
	use std::sync::Arc;

	use axum::{extract::OriginalUri, http::Uri};

	use redfish_codegen::models::odata_v4;
	use redfish_codegen::models::{
	chassis::v1_23_0::Actions as ChassisActions, chassis::v1_23_0::Chassis as ChassisModel,
	chassis::v1_23_0::ChassisType, chassis::v1_23_0::Reset as ChassisReset,
	chassis_collection::ChassisCollection as ChassisCollectionModel, resource,
	sensor::v1_7_0::Sensor as SensorModel,
	sensor_collection::SensorCollection as SensorCollectionModel,
	};

	use crate::dbus_client::{get_objects_with_interface, get_one_sensor, get_sensors_under_a_chassis};

	use crate::app_state::AppState;
	use crate::composite_query::{RequestResponse, SubRequest};

	/// Represents the state of the chassis, including a cache for chassis data.
	#[derive(Debug, Default)]
	pub struct ChassisState {
	pub cache: DashMap<String, serde_json::Value>,
	}

	/// Represents the query parameters for expanding resources.
	#[derive(serde::Deserialize, Debug, Default)]
	struct ExpandQuery {
	#[serde(rename = "$expand")]
	expand: Option<String>,
	}

	/// Retrieves all chassis from the system.
	///
	/// # Returns
	///
	/// A Result containing a vector of chassis names or an error.
	async fn get_all_chassis() -> Result<Vec<String>, Box<dyn std::error::Error>> {
	let result = get_objects_with_interface(
	Vec::new(),
	"/xyz/openbmc_project",
	"xyz.openbmc_project.Inventory.Item.Board",
	)
	.await?;
	let mut chassis: Vec<String> = Vec::new();
	for (_service, objects) in result {
	for object_path in objects {
	let path_segments: Vec<&str> = object_path.split('/').collect();
	if let Some(last_segment) = path_segments.last() {
	chassis.push(last_segment.to_string());
	}
	}
	}
	Ok(chassis)
	}

	/// Handles requests for the chassis collection.
	///
	/// # Arguments
	///
	/// * `state` - The application state.
	/// * `uri` - The original URI of the request.
	/// * `params` - Query parameters for expansion.
	async fn chassis_collection(
	State(state): State<Arc<AppState>>,
	uri: OriginalUri,
	Query(params): Query<ExpandQuery>,
	) -> impl IntoResponse {
	let params_str = params.expand.clone().unwrap_or_default();
	let cache_key = format!("{}?{}", uri.path(), params_str);
	if let Some(cached_response) = state.chassis_state.cache.get(&cache_key) {
	return (StatusCode::OK, Json(cached_response.value().clone()));
	}

	let chassis_collection = get_all_chassis().await.unwrap_or_default();
	let members = chassis_collection
	.into_iter()
	.map(\|chassis_id\| format!("{}/{}", uri.path(), chassis_id))
	.map(\|chassis_uri\| odata_v4::IdRef {
	odata_id: Some(odata_v4::Id(chassis_uri)),
	})
	.collect::<Vec<_>>();
	let response = ChassisCollectionModel {
	odata_id: odata_v4::Id(uri.path().to_string()),
	members_odata_count: odata_v4::Count(members.len() as i64),
	members,
	name: resource::Name("Chassis Collection".to_string()),
	..Default::default()
	};

	let mut response_json = serde_json::to_value(response).unwrap_or_default();
	let expand = params.expand.as_deref() == Some(".($levels=1)");
	if expand {
	let member_uris: Vec<String> = response_json["Members"]
	.as_array()
	.unwrap_or(&Vec::new())
	.iter()
	.filter_map(\|m\| m["@odata.id"].as_str().map(ToString::to_string))
	.collect();

	let tasks = member_uris
	.into_iter()
	.map(\|odata_id\| {
	let state = state.clone();
	tokio::spawn(async move {
	let uri: Uri = odata_id.parse().unwrap_or(Uri::default());
	let chassis_id = odata_id.split('/').last().unwrap_or("");
	let response = chassis(
	State(state),
	OriginalUri(uri),
	Path(chassis_id.to_string()),
	Query::default(),
	)
	.await
	.into_response();
	let body_bytes = to_bytes(response.into_body(), usize::MAX)
	.await
	.unwrap_or_default();
	serde_json::from_slice::<Value>(&body_bytes).unwrap_or_default()
	})
	})
	.collect::<Vec<_>>();
	let results = futures::future::join_all(tasks).await;

	if let Some(members) = response_json
	.get_mut("Members")
	.and_then(\|m\| m.as_array_mut())
	{
	for (member, result) in members.iter_mut().zip(results) {
	match result {
	Ok(expanded_sensor) => *member = expanded_sensor,
	Err(_e) => {
	*member = json!({"error": "Failed to expand sensor"});
	}
	}
	}
	}
	}

	// TODO: create async task to monitoring the changes related to this cache entry
	state
	.chassis_state
	.cache
	.insert(cache_key, response_json.clone());
	(StatusCode::OK, Json(response_json))
	}

	/// Handles requests for individual chassis.
	///
	/// # Arguments
	///
	/// * `state` - The application state.
	/// * `uri` - The original URI of the request.
	/// * `chassis_id` - The ID of the chassis.
	/// * `params` - Query parameters for expansion.
	async fn chassis(
	State(state): State<Arc<AppState>>,
	uri: OriginalUri,
	Path(chassis_id): Path<String>,
	Query(params): Query<ExpandQuery>,
	) -> impl IntoResponse {
	let params_str = params.expand.clone().unwrap_or_default();
	let cache_key = format!("{}?{}", uri.path(), params_str);
	if let Some(cached_response) = state.chassis_state.cache.get(&cache_key) {
	return (StatusCode::OK, Json(cached_response.value().clone()));
	}

	let response = ChassisModel {
	odata_id: odata_v4::Id(uri.path().to_string()),
	actions: Some(ChassisActions {
	chassis_reset: Some(ChassisReset {
	target: Some(format!("{}/{}", uri.path(), "Actions/Chassis.Reset")),
	..Default::default()
	}),
	..Default::default()
	}),
	sensors: Some(odata_v4::IdRef {
	odata_id: Some(odata_v4::Id(format!("{}/{}", uri.path(), "Sensors"))),
	}),
	chassis_type: ChassisType::RackMount,
	id: resource::Id(chassis_id.clone()),
	name: resource::Name(format!("Chassis {}", chassis_id)),
	..Default::default()
	};

	// TODO: create async task to monitoring the changes related to this cache entry
	let response_json = serde_json::to_value(response).unwrap_or_default();
	state
	.chassis_state
	.cache
	.insert(cache_key, response_json.clone());
	(StatusCode::OK, Json(response_json))
	}

	/// Retrieves the object path for all sensors under a chassis.
	///
	/// # Arguments
	///
	/// * `sensor_collection_uri` - The URI of the sensor collection.
	///
	/// # Returns
	///
	/// A string representing the object path for all sensors.
	fn get_all_sensors_object_path(sensor_collection_uri: &OriginalUri) -> String {
	let path = sensor_collection_uri.path().to_string();
	let parts: Vec<&str> = path.split('/').collect();
	if let Some(chassis_index) = parts.iter().position(\|&r\| r == "Chassis") {
	if let Some(chassis_name) = parts.get(chassis_index + 1) {
	return format!(
	"/xyz/openbmc_project/inventory/system/board/{}/all_sensors",
	chassis_name
	);
	}
	}
	String::new()
	}

	/// Handles requests for the sensors collection under a chassis.
	///
	/// # Arguments
	///
	/// * `state` - The application state.
	/// * `uri` - The original URI of the request.
	/// * `chassis_id` - The ID of the chassis.
	/// * `params` - Query parameters for expansion.
	async fn sensors_collection(
	State(state): State<Arc<AppState>>,
	uri: OriginalUri,
	Path(chassis_id): Path<String>,
	Query(params): Query<ExpandQuery>,
	) -> impl IntoResponse {
	let params_str = params.expand.clone().unwrap_or_default();
	let cache_key = format!("{}?{}", uri.path(), params_str);
	if let Some(cached_response) = state.chassis_state.cache.get(&cache_key) {
	return (StatusCode::OK, Json(cached_response.value().clone()));
	}

	let sensors = get_sensors_under_a_chassis(&get_all_sensors_object_path(&uri))
	.await
	.unwrap_or_default();
	let members = sensors
	.into_iter()
	.map(\|sensor_id\| format!("{}/{}", uri.path(), sensor_id))
	.map(\|sensor_uri\| odata_v4::IdRef {
	odata_id: Some(odata_v4::Id(sensor_uri)),
	})
	.collect::<Vec<_>>();

	let response = SensorCollectionModel {
	odata_id: odata_v4::Id(uri.path().to_string()),
	members_odata_count: odata_v4::Count(members.len() as i64),
	members,
	name: resource::Name("Sensor Collection".to_string()),
	..Default::default()
	};

	let mut response_json = serde_json::to_value(response).unwrap_or_default();
	let expand = params.expand.as_deref() == Some(".($levels=1)");
	if expand {
	let member_uris: Vec<String> = response_json["Members"]
	.as_array()
	.unwrap_or(&Vec::new())
	.iter()
	.filter_map(\|m\| m["@odata.id"].as_str().map(ToString::to_string))
	.collect();

	let tasks = member_uris
	.into_iter()
	.map(\|odata_id\| {
	let state = state.clone();
	let chassis_id = chassis_id.clone();
	tokio::spawn(async move {
	let uri: Uri = odata_id.parse().unwrap_or(Uri::default());
	let sensor_id = odata_id.split('/').last().unwrap_or("");
	let response = sensor(
	State(state),
	OriginalUri(uri),
	Path((chassis_id.to_string(), sensor_id.to_string())),
	Query::default(),
	)
	.await
	.into_response();
	let body_bytes = to_bytes(response.into_body(), usize::MAX)
	.await
	.unwrap_or_default();
	serde_json::from_slice::<Value>(&body_bytes).unwrap_or_default()
	})
	})
	.collect::<Vec<_>>();
	let results = futures::future::join_all(tasks).await;

	if let Some(members) = response_json
	.get_mut("Members")
	.and_then(\|m\| m.as_array_mut())
	{
	for (member, result) in members.iter_mut().zip(results) {
	match result {
	Ok(expanded_sensor) => *member = expanded_sensor,
	Err(_e) => {
	*member = json!({"error": "Failed to expand sensoris"});
	}
	}
	}
	}
	}

	// TODO: create async task to monitoring the changes related to this cache entry
	state
	.chassis_state
	.cache
	.insert(cache_key, response_json.clone());
	(StatusCode::OK, Json(response_json))
	}

	/// Formats a sensor name by removing underscores and trimming.
	///
	/// # Arguments
	///
	/// * `sensor_name` - The original sensor name.
	///
	/// # Returns
	///
	/// A formatted string representing the sensor name.
	fn format_sensor_name(sensor_name: &str) -> String {
	let trimmed_name = sensor_name
	.find('_')
	.map(\|index\| &sensor_name[index + 1..])
	.unwrap_or(sensor_name);

	trimmed_name.replace('_', " ")
	}

	/// Extracts the chassis ID from a base URL.
	///
	/// # Arguments
	///
	/// * `base_url` - The base URL containing the chassis ID.
	///
	/// # Returns
	///
	/// An Option containing the chassis ID as a String, if found.
	fn extract_chassis_id(base_url: &str) -> Option<String> {
	let segments: Vec<&str> = base_url.split('/').collect();

	segments
	.iter()
	.position(\|&segment\| segment == "Chassis")
	.and_then(\|index\| segments.get(index + 1))
	.map(\|&segment\| segment.to_string())
	}

	/// Handles a single request in the context of a composite query.
	///
	/// # Arguments
	///
	/// * `state` - The application state.
	/// * `sub_request` - The sub-request to handle.
	/// * `context` - The context for storing request results.
	///
	/// # Returns
	///
	/// A Result containing the RequestResponse or an Infallible error.
	async fn handle_one_request(
	state: &Arc<AppState>,
	sub_request: &mut SubRequest,
	context: &mut Arc<DashMap<String, Value>>,
	) -> Result<RequestResponse, Infallible> {
	let (base_url, _) = sub_request
	.url
	.split_at(sub_request.url.find('?').unwrap_or(sub_request.url.len()));
	let segments: Vec<&str> = base_url.split('/').collect();

	match sub_request.method.as_str() {
	"GET" if base_url.ends_with("/Chassis") => {
	if let Ok(parsed_url) = url::Url::parse(&format!("http://localhost{}", sub_request.url))
	{
	let query: String = parsed_url.query().unwrap_or("").to_string();
	let expanded_query: ExpandQuery = serde_qs::from_str(&query).unwrap_or_default();
	let url = sub_request.url.parse().unwrap_or_default();
	let url = OriginalUri(url);
	let response = chassis_collection(State(state.clone()), url, Query(expanded_query))
	.await
	.into_response();
	let body_bytes = to_bytes(response.into_body(), usize::MAX)
	.await
	.unwrap_or_default();
	let json = serde_json::from_slice::<Value>(&body_bytes).unwrap_or_default();
	context.insert(sub_request.referenceId.clone(), json.clone());
	Ok(RequestResponse::Json(json))
	} else {
	Ok(RequestResponse::Json(
	json!({"error": "Invalid URL format"}),
	))
	}
	}
	"GET" if segments.len() > 2 && segments[segments.len() - 2] == "Chassis" => {
	let url = sub_request.url.parse().unwrap_or_default();
	let url = OriginalUri(url);
	let chassis_id = extract_chassis_id(base_url).unwrap_or_default();
	let state = state.clone();
	let response = chassis(
	State(state),
	url,
	Path(chassis_id.to_string()),
	Query::default(),
	)
	.await
	.into_response();
	let body_bytes = to_bytes(response.into_body(), usize::MAX)
	.await
	.unwrap_or_default();
	let json = serde_json::from_slice::<Value>(&body_bytes).unwrap_or_default();
	context.insert(sub_request.referenceId.clone(), json.clone());
	Ok(RequestResponse::Json(json))
	}
	"GET" if base_url.ends_with("/Sensors") => {
	if let Ok(parsed_url) = url::Url::parse(&format!("http://localhost{}", sub_request.url))
	{
	let query: String = parsed_url.query().unwrap_or("").to_string();
	// Deserialize the query string into ExpandQuery
	let expanded_query: ExpandQuery = serde_qs::from_str(&query).unwrap_or_default();
	let url = sub_request.url.parse().unwrap_or_default();
	let url = OriginalUri(url);
	let chassis_id = extract_chassis_id(base_url).unwrap_or_default();
	let response = sensors_collection(
	State(state.clone()),
	url,
	Path(chassis_id),
	Query(expanded_query),
	)
	.await
	.into_response();
	let body_bytes = to_bytes(response.into_body(), usize::MAX)
	.await
	.unwrap_or_default();
	let json = serde_json::from_slice::<Value>(&body_bytes).unwrap_or_default();
	context.insert(sub_request.referenceId.clone(), json.clone());
	Ok(RequestResponse::Json(json))
	} else {
	Ok(RequestResponse::Json(
	json!({"error": "Invalid URL format"}),
	))
	}
	}
	"GET" if segments.len() > 2 && segments[segments.len() - 2] == "Sensors" => {
	let url = sub_request.url.parse().unwrap_or_default();
	let url = OriginalUri(url);
	let chassis_id = extract_chassis_id(base_url).unwrap_or_default();
	let sensor_id = base_url.split('/').last().unwrap_or("");
	let state = state.clone();
	let response = sensor(
	State(state),
	url,
	Path((chassis_id.to_string(), sensor_id.to_string())),
	Query::default(),
	)
	.await
	.into_response();
	let body_bytes = to_bytes(response.into_body(), usize::MAX)
	.await
	.unwrap_or_default();
	let json = serde_json::from_slice::<Value>(&body_bytes).unwrap_or_default();
	context.insert(sub_request.referenceId.clone(), json.clone());
	Ok(RequestResponse::Json(json))
	}
	_ => Ok::<RequestResponse, Infallible>(RequestResponse::Json(
	json!({"error": "Unsupported request method or URL"}),
	)),
	}
	}

	/// Retrieves a SensorModel, either from cache or by creating a new one.
	///
	/// # Arguments
	///
	/// * `state` - The application state.
	/// * `cache_key` - The key for caching the sensor model.
	/// * `odata_id` - The odata ID of the sensor.
	///
	/// # Returns
	///
	/// A Result containing the SensorModel or an Infallible error.
	async fn get_sensor_model(
	state: &Arc<AppState>,
	cache_key: &str,
	odata_id: &str,
	) -> Result<SensorModel, Infallible> {
	match state.chassis_state.cache.get(cache_key) {
	Some(cached_response) => {
	Ok(serde_json::from_value(cached_response.value().clone()).unwrap_or_default())
	}
	None => {
	let uri: Uri = odata_id.parse().unwrap_or(Uri::default());
	let uri = OriginalUri(uri);
	let sensor_id = uri.path().split('/').last().unwrap_or_default().to_string();
	let mut sensor_model = SensorModel {
	odata_id: odata_v4::Id(odata_id.to_string()),
	odata_type: odata_v4::Type("#Sensor.v1_7_0.Sensor".to_string()),
	id: resource::Id(sensor_id.to_string()),
	name: resource::Name(format_sensor_name(&sensor_id)),
	..Default::default()
	};
	get_one_sensor(&sensor_id, &mut sensor_model)
	.await
	.unwrap_or_default();
	Ok(sensor_model)
	}
	}
	}

	/// Handles requests for individual sensors.
	///
	/// # Arguments
	///
	/// * `state` - The application state.
	/// * `uri` - The original URI of the request.
	/// * `_chassis_id` - The ID of the chassis (unused).
	/// * `_sensor_id` - The ID of the sensor (unused).
	/// * `_params` - Query parameters for expansion (unused).
	async fn sensor(
	State(state): State<Arc<AppState>>,
	uri: OriginalUri,
	Path((_chassis_id, _sensor_id)): Path<(String, String)>,
	Query(_params): Query<ExpandQuery>,
	) -> impl IntoResponse {
	let cache_key = uri.path().to_string();
	let sensor_model = get_sensor_model(&state, &cache_key, &cache_key)
	.await
	.unwrap_or_default();
	let response_json = serde_json::to_value(sensor_model).unwrap_or_default();
	state
	.chassis_state
	.cache
	.insert(cache_key, response_json.clone());
	(StatusCode::OK, Json(response_json))
	}

	/// Handles an XPath segment in the context of URL expansion.
	///
	/// # Arguments
	///
	/// * `urls` - The vector of URLs to modify.
	/// * `_segment` - The XPath segment (unused).
	/// * `state` - The application state.
	///
	/// # Returns
	///
	/// A Result containing a vector of RequestResponse objects or an Infallible error.
	pub async fn handle_xpath_segment(
	urls: &mut Vec<String>,
	_segment: &str,
	state: &Arc<AppState>,
	) -> Result<Vec<RequestResponse>, Infallible> {
	let context = Arc::new(DashMap::new());
	let futures: Vec<_> = urls
	.iter()
	.map(\|url\| {
	let state_clone = state.clone();
	let mut context_clone = context.clone();
	async move {
	let mut request = SubRequest {
	method: "GET".to_string(),
	url: url.to_string(),
	..Default::default()
	};

	let response =
	handle_one_request(&state_clone, &mut request, &mut context_clone).await;
	let new_urls = match &response {
	Ok(RequestResponse::Json(json)) => {
	// TODO: use segment as query here
	let query = "$.Members[*]['@odata.id']";
	jsonpath_lib::select(json, query)
	.unwrap_or_else(\|_\| vec![])
	.iter()
	.filter_map(\|v\| v.as_str().map(String::from))
	.collect::<Vec<String>>()
	}
	_ => vec![],
	};
	(response, new_urls)
	}
	})
	.collect();

	let results = join_all(futures).await;

	*urls = vec![];
	let mut responses = vec![];
	for (response, new_urls) in results {
	urls.extend(new_urls);
	responses.push(response.unwrap_or_default());
	}

	Ok(responses)
	}

	/// Appends a segment to each URL in the provided vector.
	///
	/// # Arguments
	///
	/// * `urls` - The vector of URLs to modify.
	/// * `segment` - The segment to append to each URL.
	pub fn append_to_urls(urls: &mut [String], segment: &str) {
	urls.iter_mut().for_each(\|url\| {
	if !url.ends_with('/') && !segment.starts_with('/') {
	url.push('/');
	}
	url.push_str(segment);
	});
	}