exposing file filter for outside filestore commands

file_store/src/cli/bucket.rs

@@ -10,6 +10,7 @@ use serde::{ser::SerializeSeq, Serializer};
 use std::{
     io,
     path::{Path, PathBuf},
+    str::FromStr,
 };
 use tokio::fs;
 
@@ -48,7 +49,7 @@ impl BucketCmd {
 }
 
 #[derive(Debug, clap::Args)]
-struct FileFilter {
+pub struct FileFilter {
     /// Optional start time to look for (inclusive). Defaults to the oldest
     /// timestamp in the bucket.
     #[clap(long)]
@@ -57,15 +58,15 @@ struct FileFilter {
     /// available timestamp in the bucket.
     #[clap(long)]
     before: Option<NaiveDateTime>,
-    /// The file type to search for
+    /// The file type prefix to search for
     #[clap(long)]
-    file_type: FileType,
+    prefix: String,
 }
 
 impl FileFilter {
     fn list(&self, store: &FileStore) -> FileInfoStream {
         store.list(
-            self.file_type,
+            self.prefix.clone(),
             self.after.as_ref().map(|dt| Utc.from_utc_datetime(dt)),
             self.before.as_ref().map(|dt| Utc.from_utc_datetime(dt)),
         )
@@ -180,12 +181,15 @@ impl Locate {
     pub async fn run(&self, settings: &Settings) -> Result {
         let store = FileStore::from_settings(settings).await?;
         let file_infos = self.filter.list(&store);
-        let file_type = self.filter.file_type;
+        let prefix = self.filter.prefix.clone();
         let gateway = &self.gateway.clone();
         let mut events = store
             .source(file_infos)
             .map_ok(|buf| (buf, gateway))
-            .try_filter_map(|(buf, gateway)| async move { locate(file_type, gateway, &buf) })
+            .try_filter_map(|(buf, gateway)| {
+                let prefix = prefix.clone();
+                async move { locate(&prefix, gateway, &buf) }
+            })
             .boxed();
         let mut ser = serde_json::Serializer::new(io::stdout());
         let mut seq = ser.serialize_seq(None)?;
@@ -197,13 +201,9 @@ impl Locate {
     }
 }
 
-fn locate(
-    file_type: FileType,
-    gateway: &PublicKey,
-    buf: &[u8],
-) -> Result<Option<serde_json::Value>> {
+fn locate(prefix: &str, gateway: &PublicKey, buf: &[u8]) -> Result<Option<serde_json::Value>> {
     let pub_key = gateway.to_vec();
-    match file_type {
+    match FileType::from_str(prefix)? {
         FileType::CellHeartbeat => {
             CellHeartbeat::decode(buf).and_then(|event| event.to_value_if(pub_key))
         }

file_store/src/cli/info.rs

@@ -17,10 +17,7 @@ use helium_proto::{
     EntropyReportV1, Message, PriceReportV1,
 };
 use serde_json::json;
-use std::{
-    path::PathBuf,
-    str::FromStr,
-};
+use std::{path::PathBuf, str::FromStr};
 
 /// Print information about a given store file.
 #[derive(Debug, clap::Args)]

file_store/src/file_info.rs

@@ -66,6 +66,17 @@ impl From<(FileType, DateTime<Utc>)> for FileInfo {
     }
 }
 
+impl From<(String, DateTime<Utc>)> for FileInfo {
+    fn from(v: (String, DateTime<Utc>)) -> Self {
+        Self {
+            key: format!("{}.{}.gz", &v.0, v.1.timestamp_millis()),
+            prefix: v.0,
+            timestamp: v.1,
+            size: 0,
+        }
+    }
+}
+
 impl TryFrom<&aws_sdk_s3::model::Object> for FileInfo {
     type Error = Error;
     fn try_from(value: &aws_sdk_s3::model::Object) -> Result<Self> {

file_store/src/file_sink.rs

@@ -638,7 +638,8 @@ mod tests {
             .to_str()
             .and_then(|file_name| FileInfo::from_str(file_name).ok())
             .map_or(false, |file_info| {
-                file_info.file_type == FileType::EntropyReport
+                FileType::from_str(&file_info.prefix).expect("entropy report prefix")
+                    == FileType::EntropyReport
             })
     }
 }

file_store/src/file_store.rs

@@ -1,5 +1,5 @@
 use crate::{
-    error::DecodeError, BytesMutStream, Error, FileInfo, FileInfoStream, FileType, Result, Settings,
+    error::DecodeError, BytesMutStream, Error, FileInfo, FileInfoStream, Result, Settings,
 };
 use aws_config::meta::region::RegionProviderChain;
 use aws_sdk_s3::{types::ByteStream, Client, Endpoint, Region};
@@ -57,27 +57,27 @@ impl FileStore {
         })
     }
 
-    pub async fn list_all<A, B, F>(
+    pub async fn list_all<A, B, P>(
         &self,
-        file_type: F,
+        file_type: P,
         after: A,
         before: B,
     ) -> Result<Vec<FileInfo>>
     where
-        F: Into<FileType> + Copy,
+        P: ToString + std::fmt::Display,
         A: Into<Option<DateTime<Utc>>> + Copy,
         B: Into<Option<DateTime<Utc>>> + Copy,
     {
         self.list(file_type, after, before).try_collect().await
     }
 
-    pub fn list<A, B, F>(&self, file_type: F, after: A, before: B) -> FileInfoStream
+    pub fn list<A, B, P>(&self, prefix: P, after: A, before: B) -> FileInfoStream
     where
-        F: Into<FileType> + Copy,
+        P: ToString,
         A: Into<Option<DateTime<Utc>>> + Copy,
         B: Into<Option<DateTime<Utc>>> + Copy,
     {
-        let file_type = file_type.into();
+        let file_type = prefix.to_string();
         let before = before.into();
         let after = after.into();
 

file_store/src/lib.rs

@@ -22,6 +22,7 @@ pub mod speedtest;
 pub mod traits;
 
 pub use crate::file_store::FileStore;
+pub use cli::bucket::FileFilter;
 pub use error::{Error, Result};
 pub use file_info::{FileInfo, FileType};
 pub use file_sink::{FileSink, FileSinkBuilder};