Add Server, Filtering, and In-Memory Cache

Cargo.toml

@@ -2,4 +2,5 @@
 members = [
     "crates/bidgely_adapter",
     "crates/runner",
+    "crates/server",
 ]

crates/bidgely_adapter/src/auth.rs

@@ -1,4 +1,4 @@
-use crate::BidgelyError;
+use crate::{BidgelyError, BIDGELY_BASE_URL};
 
 #[derive(Debug, Default, Clone, serde::Serialize, serde::Deserialize)]
 #[serde(rename_all(deserialize = "camelCase"))]
@@ -8,10 +8,10 @@ pub struct UserAuthResponse {
     pub error: Option<String>,
 }
 
-pub async fn auth(base_url: &str, user_id: &str) -> Result<UserAuthResponse, BidgelyError> {
+pub async fn auth(user_id: &str) -> Result<UserAuthResponse, BidgelyError> {
     Ok(serde_json::from_str(
         &reqwest::get(format!(
-            "{base_url}/user-auth/cipher?user-id={user_id}&pilot-id=40003"
+            "{BIDGELY_BASE_URL}/user-auth/cipher?user-id={user_id}&pilot-id=40003"
         ))
         .await?
         .text()

crates/bidgely_adapter/src/feed.rs

@@ -1,7 +1,7 @@
-use crate::BidgelyError;
+use crate::{BidgelyError, BIDGELY_BASE_URL};
 use std::fs;
 
-#[derive(serde::Deserialize, serde::Serialize, Debug)]
+#[derive(Debug, serde::Deserialize, serde::Serialize)]
 #[serde(rename_all = "lowercase")]
 pub struct Feed {
     pub id: String,
@@ -11,7 +11,7 @@ pub struct Feed {
     pub entries: Vec<Entry>,
 }
 
-#[derive(serde::Deserialize, serde::Serialize, Debug)]
+#[derive(Debug, serde::Deserialize, serde::Serialize)]
 pub struct Entry {
     pub id: String,
     pub link: Vec<Link>,
@@ -21,7 +21,7 @@ pub struct Entry {
     pub updated: String,
 }
 
-#[derive(serde::Deserialize, serde::Serialize, Debug)]
+#[derive(Debug, serde::Deserialize, serde::Serialize)]
 pub struct Link {
     pub href: String,
     pub rel: String,
@@ -29,12 +29,18 @@ pub struct Link {
     pub kind: String,
 }
 
-#[derive(serde::Deserialize, serde::Serialize, Debug)]
+#[derive(Debug, serde::Deserialize, serde::Serialize)]
 pub struct Content {
     #[serde(rename(deserialize = "$value"))]
     inner: ContentType,
 }
 
+impl From<ContentType> for Content {
+    fn from(inner: ContentType) -> Self {
+        Self { inner }
+    }
+}
+
 impl Content {
     pub fn to_inner(self) -> ContentType {
         self.inner
@@ -49,7 +55,13 @@ impl std::ops::Deref for Content {
     }
 }
 
-#[derive(serde::Deserialize, serde::Serialize, Debug)]
+impl std::ops::DerefMut for Content {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.inner
+    }
+}
+
+#[derive(Debug, serde::Deserialize, serde::Serialize)]
 pub enum ContentType {
     LocalTimeParameters(LocalTimeParameters),
     UsagePoint(UsagePoint),
@@ -60,7 +72,7 @@ pub enum ContentType {
     Other,
 }
 
-#[derive(serde::Deserialize, serde::Serialize, Debug)]
+#[derive(Debug, serde::Deserialize, serde::Serialize)]
 #[serde(rename_all(deserialize = "camelCase"))]
 pub struct LocalTimeParameters {
     pub dst_end_rule: String,
@@ -69,18 +81,18 @@ pub struct LocalTimeParameters {
     pub tz_offset: String,
 }
 
-#[derive(serde::Deserialize, serde::Serialize, Debug)]
+#[derive(Debug, serde::Deserialize, serde::Serialize)]
 #[serde(rename_all(deserialize = "PascalCase"))]
 pub struct UsagePoint {
     pub service_category: ServiceCategory,
 }
 
-#[derive(serde::Deserialize, serde::Serialize, Debug)]
+#[derive(Debug, serde::Deserialize, serde::Serialize)]
 pub struct ServiceCategory {
     pub kind: u32,
 }
 
-#[derive(serde::Deserialize, serde::Serialize, Debug)]
+#[derive(Debug, serde::Deserialize, serde::Serialize)]
 #[serde(rename_all(deserialize = "camelCase"))]
 pub struct ReadingType {
     pub accumulation_behaviour: u32,
@@ -95,20 +107,20 @@ pub struct ReadingType {
     pub time_attribute: u32,
 }
 
-#[derive(serde::Deserialize, serde::Serialize, Debug)]
+#[derive(Debug, serde::Deserialize, serde::Serialize)]
 pub struct IntervalBlock {
     pub interval: Interval,
     #[serde(rename = "IntervalReading")]
     pub interval_reading: Vec<IntervalReading>,
 }
 
-#[derive(serde::Deserialize, serde::Serialize, Debug)]
+#[derive(Debug, serde::Deserialize, serde::Serialize)]
 pub struct Interval {
     pub duration: u64,
     pub start: u64,
 }
 
-#[derive(serde::Deserialize, serde::Serialize, Debug)]
+#[derive(Debug, serde::Deserialize, serde::Serialize)]
 pub struct IntervalReading {
     #[serde(rename = "ReadingQuality")]
     pub reading_quality: ReadingQuality,
@@ -117,20 +129,19 @@ pub struct IntervalReading {
     pub value: u32,
 }
 
-#[derive(serde::Deserialize, serde::Serialize, Debug)]
+#[derive(Debug, serde::Deserialize, serde::Serialize)]
 pub struct ReadingQuality {
     pub quality: u32, // todo: pub enum ReadingQuality
 }
 
-#[derive(serde::Deserialize, serde::Serialize, Debug)]
+#[derive(Debug, serde::Deserialize, serde::Serialize)]
 #[serde(rename_all(deserialize = "camelCase"))]
 pub struct TimePeriod {
-    pub duration: u32,
-    pub start: u32,
+    pub duration: u64,
+    pub start: u64,
 }
 
 pub async fn download_and_save_feed_xml(
-    base_url: &str,
     user_id: &str,
     token: &str,
     start: u64,
@@ -139,7 +150,7 @@ pub async fn download_and_save_feed_xml(
 ) -> Result<(), BidgelyError> {
     let client = reqwest::Client::new();
     let xml_data = client.get(format!(
-        "{base_url}/dashboard/users/{user_id}/gb-download?start={start}&end={end}&measurement-type=ELECTRIC"
+        "{BIDGELY_BASE_URL}/dashboard/users/{user_id}/gb-download?start={start}&end={end}&measurement-type=ELECTRIC"
     ))
         .header(reqwest::header::CONTENT_TYPE, "application/json;charset=UTF-8")
         .header(reqwest::header::AUTHORIZATION, format!("Bearer {token}"))
@@ -165,7 +176,6 @@ pub async fn download_and_save_feed_xml(
 }
 
 pub async fn get_feed(
-    base_url: &str,
     user_id: &str,
     token: &str,
     start: u64,
@@ -173,7 +183,7 @@ pub async fn get_feed(
 ) -> Result<Feed, BidgelyError> {
     let client = reqwest::Client::new();
     let xml_data = client.get(format!(
-        "{base_url}/dashboard/users/{user_id}/gb-download?start={start}&end={end}&measurement-type=ELECTRIC"
+        "{BIDGELY_BASE_URL}/dashboard/users/{user_id}/gb-download?start={start}&end={end}&measurement-type=ELECTRIC"
     ))
         .header(reqwest::header::CONTENT_TYPE, "application/json;charset=UTF-8")
         .header(reqwest::header::AUTHORIZATION, format!("Bearer {token}"))

crates/bidgely_adapter/src/lib.rs

@@ -2,6 +2,8 @@ pub mod auth;
 pub mod feed;
 pub mod session;
 
+pub(crate) const BIDGELY_BASE_URL: &'static str = "https://caapi.bidgely.com/v2.0";
+
 #[derive(thiserror::Error, Debug)]
 pub enum BidgelyError {
     #[error("reqwest error")]

crates/bidgely_adapter/src/session.rs

@@ -1,4 +1,4 @@
-use crate::BidgelyError;
+use crate::{BidgelyError, BIDGELY_BASE_URL};
 
 #[derive(Debug, Default, Clone, serde::Serialize, serde::Deserialize)]
 #[serde(rename_all(deserialize = "camelCase"))]
@@ -105,10 +105,10 @@ pub struct PremiseAddress {
     pub context: Option<String>,
 }
 
-pub async fn session(base_url: &str, session: &str) -> Result<SessionResponse, BidgelyError> {
+pub async fn session(session: &str) -> Result<SessionResponse, BidgelyError> {
     Ok(serde_json::from_str(
         &reqwest::get(format!(
-            "{base_url}/web/web-session/{session}?pilotId=40003&clientId=nsp-dashboard"
+            "{BIDGELY_BASE_URL}/web/web-session/{session}?pilotId=40003&clientId=nsp-dashboard"
         ))
         .await?
         .text()

crates/runner/Cargo.toml

@@ -6,6 +6,7 @@ edition = "2021"
 
 [dependencies]
 bidgely_adapter = { path = "../bidgely_adapter" }
+server = { path = "../server" }
 clap = { version = "4.0.17", features = ["derive"] }
 quick-xml = { version = "0.25.0", features = ["serde", "serialize"] }
 serde = { version = "1.0", features = ["derive"] }

crates/runner/src/args.rs

@@ -2,6 +2,7 @@
 pub enum Action {
     Convert,
     Download,
+    RunServer,
 }
 
 #[derive(clap::Parser, Debug)]

crates/runner/src/download.rs

@@ -1,15 +1,13 @@
-pub async fn download(base_url: &str, args: &crate::args::Args) -> Result<(), crate::error::Error> {
+pub async fn download(args: &crate::args::Args) -> Result<(), crate::error::Error> {
     println!("downloading nsp xml");
     match (&args.uid, &args.output_filename, args.start, args.end) {
         (Some(user_id), Some(output_filename), Some(start), Some(end)) => {
-            let user_auth_response = bidgely_adapter::auth::auth(base_url, user_id).await?;
+            let user_auth_response = bidgely_adapter::auth::auth(user_id).await?;
 
             let session_response =
-                bidgely_adapter::session::session(base_url, user_auth_response.payload.as_str())
-                    .await?;
+                bidgely_adapter::session::session(user_auth_response.payload.as_str()).await?;
 
             bidgely_adapter::feed::download_and_save_feed_xml(
-                base_url,
                 user_id,
                 session_response.payload.token_details.access_token.as_str(),
                 start,

crates/runner/src/error.rs

@@ -10,4 +10,6 @@ pub enum Error {
     IoError(#[from] std::io::Error),
     #[error("Bad Argument: {0}")]
     BadArgument(String),
+    #[error("Server Error")]
+    ServerError(#[from] server::error::ServerError),
 }

crates/runner/src/main.rs

@@ -6,15 +6,14 @@ pub mod convert;
 pub mod download;
 pub mod error;
 
-const BIDGELY_BASE_URL: &'static str = "https://caapi.bidgely.com/v2.0";
-
 #[tokio::main]
 async fn main() -> Result<(), error::Error> {
     let args: Args = Args::parse();
 
     match args.action {
         Action::Convert => convert::convert(&args)?,
-        Action::Download => download::download(BIDGELY_BASE_URL, &args).await?,
+        Action::Download => download::download(&args).await?,
+        Action::RunServer => server::run().await?,
     }
 
     Ok(())

crates/server/Cargo.toml

@@ -0,0 +1,16 @@
+[package]
+name = "server"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+async-trait = "0.1.58"
+bidgely_adapter = { path = "../bidgely_adapter" }
+rocket = "0.5.0-rc.2"
+rocket_contrib = { version = "0.4.11", features = ["json"] }
+serde = { version = "1.0", features = ["derive"] }
+serde_json = { version = "1.0" }
+thiserror = "1.0"
+

crates/server/src/context.rs

@@ -0,0 +1,19 @@
+use crate::nspdata::cache::in_memory_cache::InMemoryCache;
+use crate::nspdata::cache::NspCache;
+use std::sync::Arc;
+
+pub struct Context {
+    nsp_cache: Arc<dyn NspCache + Send + Sync>,
+}
+
+impl Context {
+    pub(crate) fn new_with_in_memory_cache() -> Self {
+        Self {
+            nsp_cache: Arc::new(InMemoryCache::default()),
+        }
+    }
+
+    pub(crate) fn nsp_cache(&self) -> Arc<dyn NspCache + Send + Sync> {
+        self.nsp_cache.clone()
+    }
+}

crates/server/src/error.rs

@@ -0,0 +1,29 @@
+use bidgely_adapter::session::SessionResponse;
+use std::collections::HashMap;
+use std::sync::{PoisonError, RwLockReadGuard, RwLockWriteGuard};
+
+#[derive(thiserror::Error, Debug)]
+pub enum ServerError {
+    #[error("Rocket Error {0}")]
+    RocketError(#[from] rocket::error::Error),
+    #[error("Bidgely Error {0}")]
+    BidgelyError(#[from] bidgely_adapter::BidgelyError),
+    #[error("Read Guard Poisoned")]
+    ReadGuardPoisoned,
+    #[error("Write Guard Poisoned")]
+    WriteGuardPoisoned,
+    #[error("SerdeJsonError")]
+    SerdeJsonError(#[from] serde_json::Error),
+}
+
+impl From<PoisonError<RwLockReadGuard<'_, HashMap<String, SessionResponse>>>> for ServerError {
+    fn from(_: PoisonError<RwLockReadGuard<'_, HashMap<String, SessionResponse>>>) -> Self {
+        Self::ReadGuardPoisoned
+    }
+}
+
+impl From<PoisonError<RwLockWriteGuard<'_, HashMap<String, SessionResponse>>>> for ServerError {
+    fn from(_: PoisonError<RwLockWriteGuard<'_, HashMap<String, SessionResponse>>>) -> Self {
+        Self::WriteGuardPoisoned
+    }
+}

crates/server/src/lib.rs

@@ -0,0 +1,23 @@
+#![feature(proc_macro_hygiene, decl_macro)]
+#[macro_use]
+extern crate rocket;
+
+use crate::context::Context;
+use crate::error::ServerError;
+
+pub mod error;
+
+mod context;
+mod nspdata;
+
+pub async fn run() -> Result<(), ServerError> {
+    let context = Context::new_with_in_memory_cache();
+
+    let _rocket = rocket::build()
+        .manage(context)
+        .mount("/nspdata", routes![nspdata::handle_search])
+        .launch()
+        .await?;
+
+    Ok(())
+}

crates/server/src/nspdata/cache/in_memory_cache.rs

@@ -0,0 +1,49 @@
+use crate::nspdata::cache::NspCache;
+use crate::ServerError;
+use bidgely_adapter::session::SessionResponse;
+use std::collections::HashMap;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::RwLock;
+
+#[derive(Default)]
+pub struct InMemoryCache {
+    user_id_sessions: RwLock<HashMap<String, SessionResponse>>,
+    hits: AtomicU64,
+    requests: AtomicU64,
+}
+
+#[async_trait]
+impl NspCache for InMemoryCache {
+    async fn find_session(&self, user_id: &String) -> Result<Option<SessionResponse>, ServerError> {
+        self.requests.fetch_add(1, Ordering::Relaxed);
+        let map = self.user_id_sessions.read()?;
+        Ok(match map.get(user_id).cloned() {
+            Some(session_response) => {
+                self.hits.fetch_add(1, Ordering::Relaxed);
+                Some(session_response)
+            }
+            None => None,
+        })
+    }
+
+    async fn store_session(
+        &self,
+        user_id: String,
+        session: SessionResponse,
+    ) -> Result<Option<SessionResponse>, crate::ServerError> {
+        {
+            self.user_id_sessions
+                .write()?
+                .insert(user_id.clone(), session);
+        }
+        self.find_session(&user_id).await
+    }
+
+    fn hits(&self) -> u64 {
+        self.hits.load(Ordering::Relaxed)
+    }
+
+    fn requests(&self) -> u64 {
+        self.requests.load(Ordering::Relaxed)
+    }
+}

crates/server/src/nspdata/cache/mod.rs

@@ -0,0 +1,18 @@
+use bidgely_adapter::session::SessionResponse;
+
+pub(crate) mod in_memory_cache;
+
+#[async_trait]
+pub(crate) trait NspCache {
+    async fn find_session(
+        &self,
+        user_id: &String,
+    ) -> Result<Option<SessionResponse>, crate::ServerError>;
+    async fn store_session(
+        &self,
+        user_id: String,
+        session: SessionResponse,
+    ) -> Result<Option<SessionResponse>, crate::ServerError>;
+    fn hits(&self) -> u64;
+    fn requests(&self) -> u64;
+}

crates/server/src/nspdata/mod.rs

@@ -0,0 +1,16 @@
+pub(crate) mod cache;
+pub(crate) mod search;
+
+use crate::Context;
+use rocket::State;
+
+#[get("/<user_id>?<start>&<end>")]
+pub(crate) async fn handle_search(
+    context: &State<Context>,
+    user_id: String,
+    start: u64,
+    end: u64,
+) -> String {
+    // todo: impl Responder for ServerError to eliminate these unwraps
+    serde_json::to_string(&search::handler(context, user_id, start, end).await.unwrap()).unwrap()
+}

crates/server/src/nspdata/search.rs

@@ -0,0 +1,96 @@
+use crate::{Context, ServerError};
+use bidgely_adapter::feed::{ContentType, IntervalReading};
+use bidgely_adapter::session::{SessionResponse, UserProfileDetails};
+use rocket::State;
+use std::ops::DerefMut;
+
+#[derive(Debug, serde::Serialize, serde::Deserialize)]
+pub(crate) struct SearchResponse {
+    user_profile: UserProfileDetails,
+    readings: Vec<PrettyIntervalReading>,
+}
+
+#[derive(Debug, serde::Serialize, serde::Deserialize)]
+pub(crate) struct PrettyIntervalReading {
+    start: u64,
+    duration: u64,
+    quality: u32,
+    value: u32,
+}
+
+impl From<(SessionResponse, Vec<IntervalReading>)> for SearchResponse {
+    fn from((session_response, readings): (SessionResponse, Vec<IntervalReading>)) -> Self {
+        Self {
+            user_profile: session_response.payload.user_profile_details,
+            readings: readings
+                .into_iter()
+                .map(|reading| PrettyIntervalReading {
+                    start: reading.time_period.start,
+                    duration: reading.time_period.duration,
+                    quality: reading.reading_quality.quality,
+                    value: reading.value,
+                })
+                .collect::<Vec<PrettyIntervalReading>>(),
+        }
+    }
+}
+
+pub(crate) async fn handler(
+    context: &State<Context>,
+    user_id: String,
+    start: u64,
+    end: u64,
+) -> Result<SearchResponse, crate::error::ServerError> {
+    let session_response = get_session(context, user_id.clone()).await?;
+
+    let mut feed = bidgely_adapter::feed::get_feed(
+        user_id.as_str(),
+        session_response.payload.token_details.access_token.as_str(),
+        start,
+        end,
+    )
+    .await?;
+
+    // todo: think if there is a nicer way to do this via some sort of filter / retain / map / collect as opposed
+    // to creating a new vector and moving elements into it.
+    let mut return_data: Vec<IntervalReading> = vec![];
+
+    feed.entries
+        .iter_mut()
+        .for_each(|entry| match entry.content.deref_mut() {
+            ContentType::IntervalBlock(interval_block) => {
+                interval_block.interval_reading.retain(|interval_reading| {
+                    (interval_reading.time_period.start > start)
+                        && ((interval_reading.time_period.start
+                            + interval_reading.time_period.duration)
+                            <= end)
+                });
+                return_data.append(&mut interval_block.interval_reading)
+            }
+            _ => {}
+        });
+
+    Ok((session_response, return_data).into())
+}
+
+async fn get_session(
+    context: &State<Context>,
+    user_id: String,
+) -> Result<SessionResponse, ServerError> {
+    let cache = context.nsp_cache();
+
+    Ok(match cache.find_session(&user_id).await? {
+        Some(session) => session,
+        None => {
+            let user_auth_response = bidgely_adapter::auth::auth(user_id.as_str()).await.unwrap();
+
+            let session_response =
+                bidgely_adapter::session::session(user_auth_response.payload.as_str()).await?;
+
+            cache
+                .store_session(user_id.clone(), session_response)
+                .await?
+                .unwrap()
+        }
+    })
+}