Remove FileStore struct from file_store (#1048)

* Remove FileStore struct from file_store

* Remove region from settings, just inherit region from environment

* move all low level functions to lib.rs

* Fix aws_local

* fix PriceTracker

* add lock file

* fix price server

* Update ingest

* Added reexported Client type from file-store

* update price to use file_store::Client

* updateig mobile packet verifier

* update mobile-verifier

* update reward-indexer

* update iot packet verifier

* update poc_entropy and iot_verifier

* update aws_local to file_store_client

* fixing tests and clippy

* run formatter

* remove commented out code

* update file_info_poller builder to take Into<String> for bucket

* update generics on file_info_poller and minor refactor of file uploader

* update bucket args to be impl Into<String> in file_store/src/lib.rs

* refactor to file_store_client

Author: bbalser

aws_local/src/lib.rs

@@ -1,10 +1,9 @@
 use anyhow::{anyhow, Result};
-use aws_config::meta::region::RegionProviderChain;
-use aws_config::{BehaviorVersion, Region};
+use aws_config::BehaviorVersion;
 use aws_sdk_s3::Client;
 use chrono::Utc;
 use file_store::traits::MsgBytes;
-use file_store::{file_sink, file_upload, FileStore, FileType, Settings};
+use file_store::{file_sink, file_upload, FileType, Settings};
 use std::env;
 use std::path::Path;
 use std::sync::Arc;
@@ -27,22 +26,17 @@ pub fn gen_bucket_name() -> String {
 // Used to create mocked aws buckets and files.
 pub struct AwsLocal {
     pub fs_settings: Settings,
-    pub file_store: FileStore,
-    pub aws_client: aws_sdk_s3::Client,
+    pub file_store_client: file_store::Client,
+    bucket: String,
 }
 
 impl AwsLocal {
     async fn create_aws_client(settings: &Settings) -> aws_sdk_s3::Client {
-        let region = Region::new(settings.region.clone());
-        let region_provider = RegionProviderChain::first_try(region).or_default_provider();
-
-        let config = aws_config::defaults(BehaviorVersion::latest())
-            .region(region_provider)
-            .load()
-            .await;
+        let config = aws_config::defaults(BehaviorVersion::latest()).load().await;
 
         let mut s3_config = aws_sdk_s3::config::Builder::from(&config)
             .force_path_style(true)
+            .region(aws_config::Region::new("us-east-1"))
             .endpoint_url(settings.endpoint.as_ref().expect("endpoint"));
 
         let creds = aws_sdk_s3::config::Credentials::builder()
@@ -62,20 +56,16 @@ impl AwsLocal {
 
     pub async fn new(endpoint: &str, bucket: &str) -> AwsLocal {
         let settings = Settings {
-            bucket: bucket.into(),
             endpoint: Some(endpoint.into()),
-            region: "us-east-1".into(),
             access_key_id: Some("random".into()),
             secret_access_key: Some("random2".into()),
         };
         let client = Self::create_aws_client(&settings).await;
         client.create_bucket().bucket(bucket).send().await.unwrap();
         AwsLocal {
-            aws_client: client,
+            file_store_client: client,
             fs_settings: settings.clone(),
-            file_store: file_store::FileStore::from_settings(&settings)
-                .await
-                .unwrap(),
+            bucket: bucket.to_string(),
         }
     }
 
@@ -95,9 +85,7 @@ impl AwsLocal {
         let (shutdown_trigger, shutdown_listener) = triggered::trigger();
 
         let (file_upload, file_upload_server) =
-            file_upload::FileUpload::from_settings_tm(&self.fs_settings)
-                .await
-                .unwrap();
+            file_upload::FileUpload::new(self.file_store_client.clone(), self.bucket.clone()).await;
 
         let (item_sink, item_server) =
             file_sink::FileSinkBuilder::new(file_type, &tmp_dir_path, file_upload, metric_name)
@@ -158,7 +146,7 @@ impl AwsLocal {
         if !file_path.is_file() {
             return Err(anyhow!("File {path_str} is not a file"));
         }
-        self.file_store.put(file_path).await?;
+        file_store::put_file(&self.file_store_client, &self.bucket, file_path).await?;
 
         Ok(())
     }

file_store/src/cli/bucket.rs

@@ -6,7 +6,7 @@ use crate::{
     mobile_radio_threshold::VerifiedRadioThresholdIngestReport,
     speedtest::{cli::SpeedtestAverage, CellSpeedtest},
     traits::MsgDecode,
-    Error, FileInfoStream, FileStore, FileType, Result, Settings,
+    Error, FileInfoStream, FileType, Result, Settings,
 };
 use chrono::{NaiveDateTime, TimeZone, Utc};
 use futures::{stream::TryStreamExt, StreamExt, TryFutureExt};
@@ -73,8 +73,10 @@ pub struct FileFilter {
 }
 
 impl FileFilter {
-    pub fn list(&self, store: &FileStore) -> FileInfoStream {
-        store.list(
+    pub fn list(&self, client: &aws_sdk_s3::Client, bucket: &str) -> FileInfoStream {
+        crate::list_files(
+            client,
+            bucket,
             &self.prefix,
             self.after.as_ref().map(|dt| Utc.from_utc_datetime(dt)),
             self.before.as_ref().map(|dt| Utc.from_utc_datetime(dt)),
@@ -85,37 +87,40 @@ impl FileFilter {
 /// List keys in a given bucket
 #[derive(Debug, clap::Args)]
 pub struct List {
+    #[clap(short)]
+    bucket: String,
     #[clap(flatten)]
     filter: FileFilter,
 }
 
 impl List {
     pub async fn run(&self, settings: &Settings) -> Result {
-        let store = FileStore::from_settings(settings).await?;
-        let mut file_infos = self.filter.list(&store);
+        let client = settings.connect().await;
+        let mut file_infos = self.filter.list(&client, &self.bucket);
         let mut ser = serde_json::Serializer::new(io::stdout());
         let mut seq = ser.serialize_seq(None)?;
         while let Some(info) = file_infos.try_next().await? {
             seq.serialize_element(&info)?;
         }
         seq.end()?;
         Ok(())
-        // print_json(&file_infos)
     }
 }
 
 /// Put one or more files in a given bucket
 #[derive(Debug, clap::Args)]
 pub struct Put {
+    #[clap(short)]
+    bucket: String,
     /// The files to upload to the bucket
     files: Vec<PathBuf>,
 }
 
 impl Put {
     pub async fn run(&self, settings: &Settings) -> Result {
-        let file_store = FileStore::from_settings(settings).await?;
+        let client = settings.connect().await;
         for file in self.files.iter() {
-            file_store.put(file).await?;
+            crate::put_file(&client, &self.bucket, file).await?;
         }
         Ok(())
     }
@@ -132,9 +137,9 @@ pub struct Remove {
 
 impl Remove {
     pub async fn run(&self, settings: &Settings) -> Result {
-        let file_store = FileStore::from_settings(settings).await?;
+        let client = settings.connect().await;
         for key in self.keys.iter() {
-            file_store.remove(key).await?;
+            crate::remove_file(&client, &self.bucket, key).await?;
         }
         Ok(())
     }
@@ -143,6 +148,7 @@ impl Remove {
 /// Get one or more files from a given bucket to a given folder
 #[derive(Debug, clap::Args)]
 pub struct Get {
+    bucket: String,
     /// The target folder to download files to
     dest: PathBuf,
     #[clap(flatten)]
@@ -151,24 +157,27 @@ pub struct Get {
 
 impl Get {
     pub async fn run(&self, settings: &Settings) -> Result {
-        let store = FileStore::from_settings(settings).await?;
-        let file_infos = self.filter.list(&store);
+        let client = settings.connect().await;
+        let file_infos = self.filter.list(&client, &self.bucket);
         file_infos
-            .map_ok(|info| (store.clone(), info))
-            .try_for_each_concurrent(5, |(store, info)| async move {
+            .map_ok(|info| (client.clone(), self.bucket.clone(), info))
+            .try_for_each_concurrent(5, |(client, bucket, info)| async move {
                 fs::OpenOptions::new()
                     .write(true)
                     .create(true)
                     .truncate(true)
                     .open(&self.dest.join(Path::new(&info.key)))
                     .map_err(Error::from)
                     .and_then(|mut file| {
-                        store.get_raw(&info.key).and_then(|stream| async move {
-                            let mut reader = tokio::io::BufReader::new(stream.into_async_read());
-                            tokio::io::copy(&mut reader, &mut file)
-                                .map_err(Error::from)
-                                .await
-                        })
+                        crate::get_raw_file(&client, &bucket, &info.key).and_then(
+                            |stream| async move {
+                                let mut reader =
+                                    tokio::io::BufReader::new(stream.into_async_read());
+                                tokio::io::copy(&mut reader, &mut file)
+                                    .map_err(Error::from)
+                                    .await
+                            },
+                        )
                     })
                     .map_ok(|_| ())
                     .await
@@ -181,6 +190,7 @@ impl Get {
 /// Locate specific records in a time range
 #[derive(Debug, clap::Args)]
 pub struct Locate {
+    bucket: String,
     gateway: PublicKey,
 
     #[clap(flatten)]
@@ -189,12 +199,11 @@ pub struct Locate {
 
 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 client = settings.connect().await;
+        let file_infos = self.filter.list(&client, &self.bucket);
         let prefix = self.filter.prefix.clone();
         let gateway = &self.gateway.clone();
-        let mut events = store
-            .source(file_infos)
+        let mut events = crate::source_files(&client, &self.bucket, file_infos)
             .map_ok(|buf| (buf, gateway))
             .try_filter_map(|(buf, gateway)| {
                 let prefix = prefix.clone();

file_store/src/file_info_poller.rs

@@ -1,4 +1,4 @@
-use crate::{file_store, traits::MsgDecode, Error, FileInfo, FileStore, Result};
+use crate::{traits::MsgDecode, Error, FileInfo, Result};
 use aws_sdk_s3::primitives::ByteStream;
 use chrono::{DateTime, Utc};
 use derive_builder::Builder;
@@ -130,11 +130,19 @@ pub struct FileInfoPollerConfig<Message, State, Store, Parser> {
     p: PhantomData<Message>,
 }
 
+impl<Message, State, Parser>
+    FileInfoPollerConfigBuilder<Message, State, FileStoreInfoPollerStore, Parser>
+{
+    pub fn file_store(self, client: crate::Client, bucket: impl Into<String>) -> Self {
+        self.store(FileStoreInfoPollerStore::new(client, bucket))
+    }
+}
+
 #[derive(Clone)]
 pub struct FileInfoPollerServer<
     Message,
     State,
-    Store = FileStore,
+    Store = FileStoreInfoPollerStore,
     Parser = MsgDecodeFileInfoPollerParser,
 > {
     config: FileInfoPollerConfig<Message, State, Store, Parser>,
@@ -179,11 +187,13 @@ where
     }
 }
 
-impl<Message, State, Parser> ManagedTask for FileInfoPollerServer<Message, State, FileStore, Parser>
+impl<Message, State, Store, Parser> ManagedTask
+    for FileInfoPollerServer<Message, State, Store, Parser>
 where
     Message: Send + Sync + 'static,
     State: FileInfoPollerState,
     Parser: FileInfoPollerParser<Message>,
+    Store: FileInfoPollerStore,
 {
     fn start_task(
         self: Box<Self>,
@@ -339,7 +349,7 @@ where
     T: MsgDecode + TryFrom<T::Msg, Error = Error> + Send + Sync + 'static,
 {
     async fn parse(&self, byte_stream: ByteStream) -> Result<Vec<T>> {
-        Ok(file_store::stream_source(byte_stream)
+        Ok(crate::stream_source(byte_stream)
             .filter_map(|msg| async {
                 msg.map_err(|err| {
                     tracing::error!(
@@ -374,7 +384,7 @@ where
     T: helium_proto::Message + Default,
 {
     async fn parse(&self, byte_stream: ByteStream) -> Result<Vec<T>> {
-        Ok(file_store::stream_source(byte_stream)
+        Ok(crate::stream_source(byte_stream)
             .filter_map(|msg| async {
                 msg.map_err(|err| {
                     tracing::error!(
@@ -409,21 +419,35 @@ async fn cache_file(cache: &MemoryFileCache, file_info: &FileInfo) {
     cache.insert(file_info.key.clone(), true, CACHE_TTL).await;
 }
 
+pub struct FileStoreInfoPollerStore {
+    client: crate::Client,
+    bucket: String,
+}
+
+impl FileStoreInfoPollerStore {
+    fn new(client: crate::Client, bucket: impl Into<String>) -> Self {
+        Self {
+            client,
+            bucket: bucket.into(),
+        }
+    }
+}
+
 #[async_trait::async_trait]
-impl FileInfoPollerStore for FileStore {
+impl FileInfoPollerStore for FileStoreInfoPollerStore {
     async fn list_all<A, B>(&self, file_type: &str, after: A, before: B) -> Result<Vec<FileInfo>>
     where
         A: Into<Option<DateTime<Utc>>> + Send + Sync + Copy,
         B: Into<Option<DateTime<Utc>>> + Send + Sync + Copy,
     {
-        self.list_all(file_type, after, before).await
+        crate::list_all_files(&self.client, &self.bucket, file_type, after, before).await
     }
 
     async fn get_raw<K>(&self, key: K) -> Result<ByteStream>
     where
         K: Into<String> + Send + Sync,
     {
-        self.get_raw(key).await
+        crate::get_raw_file(&self.client, &self.bucket, key).await
     }
 }