feat: added API to fetch a single event by ID (#268)

Author: ErikBjare

.github/workflows/build.yml

@@ -73,7 +73,7 @@ jobs:
         rustup component add llvm-tools-preview
     - name: Download grcov
       run: |
-        curl -L https://github.com/mozilla/grcov/releases/latest/download/grcov-linux-x86_64.tar.bz2 | tar jxf -
+        curl -L https://github.com/mozilla/grcov/releases/latest/download/grcov-x86_64-unknown-linux-gnu.tar.bz2  | tar jxf -
     - name: Run tests with coverage
       run: |
         # Add cwd to path to find grcov

aw-datastore/src/datastore.rs

@@ -662,6 +662,67 @@ impl DatastoreInstance {
         Ok(inserted_heartbeat)
     }
 
+    pub fn get_event(
+        &mut self,
+        conn: &Connection,
+        bucket_id: &str,
+        event_id: i64,
+    ) -> Result<Event, DatastoreError> {
+        let bucket = self.get_bucket(bucket_id)?;
+
+        let mut stmt = match conn.prepare(
+            "
+                SELECT id, starttime, endtime, data
+                FROM events
+                WHERE bucketrow = ?1
+                    AND id = ?2
+                LIMIT 1
+            ;",
+        ) {
+            Ok(stmt) => stmt,
+            Err(err) => {
+                return Err(DatastoreError::InternalError(format!(
+                    "Failed to prepare get_event SQL statement: {}",
+                    err
+                )))
+            }
+        };
+
+        // TODO: Refactor to share row-parsing logic with get_events
+        let row = match stmt.query_row(&[&bucket.bid.unwrap(), &event_id], |row| {
+            let id = row.get(0)?;
+            let starttime_ns: i64 = row.get(1)?;
+            let endtime_ns: i64 = row.get(2)?;
+            let data_str: String = row.get(3)?;
+
+            let time_seconds: i64 = (starttime_ns / 1_000_000_000) as i64;
+            let time_subnanos: u32 = (starttime_ns % 1_000_000_000) as u32;
+            let duration_ns = endtime_ns - starttime_ns;
+            let data: serde_json::map::Map<String, Value> =
+                serde_json::from_str(&data_str).unwrap();
+
+            Ok(Event {
+                id: Some(id),
+                timestamp: DateTime::<Utc>::from_utc(
+                    NaiveDateTime::from_timestamp(time_seconds, time_subnanos),
+                    Utc,
+                ),
+                duration: Duration::nanoseconds(duration_ns),
+                data,
+            })
+        }) {
+            Ok(rows) => rows,
+            Err(err) => {
+                return Err(DatastoreError::InternalError(format!(
+                    "Failed to map get_event SQL statement: {}",
+                    err
+                )))
+            }
+        };
+
+        Ok(row)
+    }
+
     pub fn get_events(
         &mut self,
         conn: &Connection,

aw-datastore/src/worker.rs

@@ -64,6 +64,7 @@ pub enum Command {
     GetBuckets(),
     InsertEvents(String, Vec<Event>),
     Heartbeat(String, Event, f64),
+    GetEvent(String, i64),
     GetEvents(
         String,
         Option<DateTime<Utc>>,
@@ -241,6 +242,12 @@ impl DatastoreWorker {
                     Err(e) => Err(e),
                 }
             }
+            Command::GetEvent(bucketname, event_id) => {
+                match ds.get_event(&transaction, &bucketname, event_id) {
+                    Ok(el) => Ok(Response::Event(el)),
+                    Err(e) => Err(e),
+                }
+            }
             Command::GetEvents(bucketname, starttime_opt, endtime_opt, limit_opt) => {
                 match ds.get_events(
                     &transaction,
@@ -395,6 +402,18 @@ impl Datastore {
         }
     }
 
+    pub fn get_event(&self, bucket_id: &str, event_id: i64) -> Result<Event, DatastoreError> {
+        let cmd = Command::GetEvent(bucket_id.to_string(), event_id);
+        let receiver = self.requester.request(cmd).unwrap();
+        match receiver.collect().unwrap() {
+            Ok(r) => match r {
+                Response::Event(el) => Ok(el),
+                _ => panic!("Invalid response"),
+            },
+            Err(e) => Err(e),
+        }
+    }
+
     pub fn get_events(
         &self,
         bucket_id: &str,

aw-datastore/tests/datastore.rs

@@ -121,6 +121,34 @@ mod datastore_tests {
         }
     }
 
+    #[test]
+    fn test_events_get_single() {
+        // Setup datastore
+        let ds = Datastore::new_in_memory(false);
+        let bucket = create_test_bucket(&ds);
+
+        // Insert event
+        let e1 = Event {
+            id: None,
+            timestamp: Utc::now(),
+            duration: Duration::seconds(0),
+            data: json_map! {"key": json!("value")},
+        };
+        let mut e2 = e1.clone();
+        e2.timestamp = e2.timestamp + Duration::nanoseconds(1);
+
+        let event_list = [e1.clone(), e2.clone()];
+        ds.insert_events(&bucket.id, &event_list).unwrap();
+
+        let events = ds.get_events(&bucket.id, None, None, None).unwrap();
+        let first_event = events.first().unwrap();
+        let first_event_id = first_event.id.unwrap();
+
+        let fetched_event = ds.get_event(&bucket.id, first_event_id).unwrap();
+        // TODO: Check entire events to ensure integrity
+        assert_eq!(fetched_event.id.unwrap(), first_event_id);
+    }
+
     #[test]
     fn test_events_get_filters() {
         // Setup datastore

aw-server/src/endpoints/bucket.rs

@@ -101,6 +101,23 @@ pub fn bucket_events_get(
     }
 }
 
+// Needs unused parameter, otherwise there'll be a route collision
+// See: https://api.rocket.rs/master/rocket/struct.Route.html#resolving-collisions
+#[get("/<bucket_id>/events/<event_id>?<_unused..>")]
+pub fn bucket_events_get_single(
+    bucket_id: String,
+    event_id: i64,
+    _unused: Option<u64>,
+    state: &State<ServerState>,
+) -> Result<Json<Event>, HttpErrorJson> {
+    let datastore = endpoints_get_lock!(state.datastore);
+    let res = datastore.get_event(&bucket_id, event_id);
+    match res {
+        Ok(events) => Ok(Json(events)),
+        Err(err) => Err(err.into()),
+    }
+}
+
 #[post("/<bucket_id>/events", data = "<events>", format = "application/json")]
 pub fn bucket_events_create(
     bucket_id: String,

aw-server/src/endpoints/mod.rs

@@ -121,6 +121,7 @@ pub fn build_rocket(server_state: ServerState, config: AWConfig) -> rocket::Rock
                 bucket::bucket_events_create,
                 bucket::bucket_events_heartbeat,
                 bucket::bucket_event_count,
+                bucket::bucket_events_get_single,
                 bucket::bucket_events_delete_by_id,
                 bucket::bucket_export
             ],