Add key item cache for garmin sync (#45)

* keep track of files in key_item_cache table

.github/workflows/rust.yml

@@ -32,16 +32,16 @@ jobs:
     - name: Clippy
       run: cargo clippy -- -W clippy::pedantic
 
-    - name: Outdated
-      run: |
-        cargo install cargo-outdated && \
-        cargo outdated -d2
+    # - name: Outdated
+    #   run: |
+    #     cargo install cargo-outdated && \
+    #     cargo outdated -d2
 
-    - name: Unused Deps
-      run: |
-        rustup update nightly && \
-        cargo +nightly install cargo-udeps && \
-        cargo +nightly udeps
+    # - name: Unused Deps
+    #   run: |
+    #     rustup update nightly && \
+    #     cargo +nightly install cargo-udeps && \
+    #     cargo +nightly udeps
 
     - name: Build
       run: |

fitbit_bot/src/telegram_bot.rs

@@ -73,7 +73,7 @@ impl TelegramBot {
             )
             .await
             {
-                Err(_) | Ok(Ok(_)) => FAILURE_COUNT.reset()?,
+                Err(_) | Ok(Ok(())) => FAILURE_COUNT.reset()?,
                 Ok(Err(_)) => FAILURE_COUNT.increment()?,
             }
         }

fitbit_lib/Cargo.toml

@@ -18,7 +18,7 @@ rayon = "1.5"
 log = "0.4"
 lazy_static = "1.4"
 parking_lot = "0.12"
-itertools = "0.11"
+itertools = "0.12"
 avro-rs = {version = "0.13", features = ["snappy"]}
 uuid = { version = "1.0", features = ["serde", "v4"] }
 postgres_query = {git = "https://github.com/ddboline/rust-postgres-query", tag = "0.3.5", features=["deadpool"]}
@@ -36,7 +36,7 @@ fitparser = {git="https://github.com/ddboline/fitparse-rs.git", branch="time-0.3
 smallvec = "1.6"
 crossbeam-utils = "0.8"
 derive_more = "0.99"
-polars = {version="0.33", features=["temporal", "parquet", "lazy"]}
+polars = {version="0.35", features=["temporal", "parquet", "lazy"]}
 stack-string = { git = "https://github.com/ddboline/stack-string-rs.git", features=["postgres_types"], tag="0.9.2" }
 
 [dev-dependencies]

garmin_cli/Cargo.toml

@@ -26,10 +26,10 @@ rayon = "1.5"
 regex = "1.4"
 tempdir = "0.3"
 serde_json = "1.0"
-itertools = "0.11"
+itertools = "0.12"
 smallvec = "1.6"
 tokio-postgres = {version = "0.7", features = ["with-time-0_3"]}
 refinery = {version="0.8", features=["tokio-postgres"]}
 stack-string = { git = "https://github.com/ddboline/stack-string-rs.git", features=["postgres_types"], tag="0.9.2" }
 stdout-channel = "0.6"
-aws-config = "0.56"
+aws-config = {version="0.101", features=["behavior-version-latest"]}

garmin_cli/src/garmin_cli.rs

@@ -44,7 +44,7 @@ use garmin_reports::{
 
 #[derive(Debug, PartialEq, Clone, Eq)]
 pub enum GarminCliOptions {
-    Sync(bool),
+    Sync,
     All,
     Bootstrap,
     FileNames(Vec<PathBuf>),
@@ -143,7 +143,7 @@ impl GarminCli {
             let pool = self.get_pool();
             GarminSummary::write_summary_to_postgres(&summary_list, &pool)
                 .await
-                .map(|_| Vec::new())
+                .map(|()| Vec::new())
         }
     }
 
@@ -219,12 +219,12 @@ impl GarminCli {
     /// # Errors
     /// Return error if `sync_everything` fails
     pub async fn run_bootstrap(&self) -> Result<Vec<StackString>, Error> {
-        self.sync_everything(true).await
+        self.sync_everything().await
     }
 
     /// # Errors
     /// Return error if `sync_dir` fails
-    pub async fn sync_everything(&self, check_md5: bool) -> Result<Vec<StackString>, Error> {
+    pub async fn sync_everything(&self) -> Result<Vec<StackString>, Error> {
         let sdk_config = aws_config::load_from_env().await;
         let gsync = GarminSync::new(&sdk_config);
 
@@ -233,34 +233,30 @@ impl GarminCli {
                 "Syncing GPS files",
                 &self.get_config().gps_dir,
                 &self.get_config().gps_bucket,
-                check_md5,
             ),
             (
                 "Syncing CACHE files",
                 &self.get_config().cache_dir,
                 &self.get_config().cache_bucket,
-                check_md5,
             ),
             (
                 "Syncing Fitbit Cache",
                 &self.get_config().fitbit_cachedir,
                 &self.get_config().fitbit_bucket,
-                check_md5,
             ),
             (
                 "Syncing Fitbit Archive",
                 &self.get_config().fitbit_archivedir,
                 &self.get_config().fitbit_archive_bucket,
-                check_md5,
             ),
         ];
 
-        let futures = options
-            .into_iter()
-            .map(|(title, local_dir, s3_bucket, check_md5)| {
-                debug!("{}", title);
-                gsync.sync_dir(title, local_dir, s3_bucket, check_md5)
-            });
+        let futures = options.into_iter().map(|(title, local_dir, s3_bucket)| {
+            debug!("{}", title);
+            let pool = self.pool.clone();
+            let gsync = gsync.clone();
+            async move { gsync.sync_dir(title, local_dir, s3_bucket, &pool).await }
+        });
         try_join_all(futures).await
     }
 

garmin_cli/src/garmin_cli_opts.rs

@@ -86,10 +86,7 @@ pub enum GarminCliOpts {
         #[clap(short, long)]
         end_date: Option<DateType>,
     },
-    Sync {
-        #[clap(short, long)]
-        md5sum: bool,
-    },
+    Sync,
     #[clap(alias = "fit")]
     Fitbit {
         #[clap(short, long)]
@@ -160,7 +157,7 @@ impl GarminCliOpts {
             .process_opts(&config)
             .await?;
             Self::Strava.process_opts(&config).await?;
-            Self::Sync { md5sum: false }.process_opts(&config).await
+            Self::Sync.process_opts(&config).await
         } else {
             opts.process_opts(&config).await
         }
@@ -191,7 +188,7 @@ impl GarminCliOpts {
                 start_date: start_date.map(Into::into),
                 end_date: end_date.map(Into::into),
             },
-            Self::Sync { md5sum } => GarminCliOptions::Sync(md5sum),
+            Self::Sync => GarminCliOptions::Sync,
             Self::SyncAll => {
                 return Ok(());
             }
@@ -225,7 +222,7 @@ impl GarminCliOpts {
                 let filenames = client.sync_tcx(start_date).await?;
                 if !filenames.is_empty() {
                     let mut buf = cli.proc_everything().await?;
-                    buf.extend_from_slice(&cli.sync_everything(false).await?);
+                    buf.extend_from_slice(&cli.sync_everything().await?);
                 }
                 let filenames = filenames
                     .into_iter()
@@ -473,7 +470,7 @@ impl GarminCliOpts {
                 cli.proc_everything().await
             }
             Some(GarminCliOptions::Bootstrap) => cli.run_bootstrap().await,
-            Some(GarminCliOptions::Sync(check_md5)) => cli.sync_everything(*check_md5).await,
+            Some(GarminCliOptions::Sync) => cli.sync_everything().await,
             Some(GarminCliOptions::Connect {
                 data_directory,
                 start_date,
@@ -484,7 +481,7 @@ impl GarminCliOpts {
                     Self::sync_with_garmin_connect(cli, data_directory, *start_date, *end_date)
                         .await?;
                 if !filenames.is_empty() || !input_files.is_empty() || !dates.is_empty() {
-                    buf.extend_from_slice(&cli.sync_everything(false).await?);
+                    buf.extend_from_slice(&cli.sync_everything().await?);
                     if let Ok(client) = FitbitClient::with_auth(cli.config.clone()).await {
                         let result = client.sync_everything(&cli.pool).await?;
                         buf.push(format_sstr!(

garmin_connect_lib/Cargo.toml

@@ -24,6 +24,6 @@ hyper = "0.14"
 derive_more = "0.99"
 bytes = "1.0"
 smallvec = "1.6"
-itertools = "0.11"
+itertools = "0.12"
 http = "0.2"
 stack-string = { git = "https://github.com/ddboline/stack-string-rs.git", features=["postgres_types"], tag="0.9.2" }

garmin_http/Cargo.toml

@@ -11,7 +11,6 @@ garmin_lib = {path = "../garmin_lib"}
 garmin_cli = {path = "../garmin_cli"}
 garmin_reports = {path="../garmin_reports"}
 race_result_analysis = {path="../race_result_analysis"}
-http = "0.2"
 tokio = {version="1.32", features=["full"]}
 tokio-stream = "0.1"
 time = {version="0.3", features=["serde-human-readable", "macros", "formatting", "parsing"]}
@@ -29,8 +28,8 @@ thiserror = "1.0"
 async-trait = "0.1"
 rayon = "1.5"
 handlebars = "4.0"
-itertools = "0.11"
-cookie = {version="0.17", features=["percent-encode"]}
+itertools = "0.12"
+cookie = {version="0.18", features=["percent-encode"]}
 base64 = "0.21"
 url = "2.3"
 serde_yaml = "0.9"

garmin_http/src/errors.rs

@@ -1,7 +1,7 @@
 use anyhow::Error as AnyhowError;
 use base64::DecodeError;
 use handlebars::RenderError;
-use http::StatusCode;
+use rweb::http::StatusCode;
 use log::error;
 use postgres_query::Error as PqError;
 use rweb::{

garmin_http/src/garmin_requests.rs

@@ -80,7 +80,7 @@ impl StravaSyncRequest {
 
         if !filenames.is_empty() {
             gcli.process_filenames(&filenames).await?;
-            gcli.sync_everything(false).await?;
+            gcli.sync_everything().await?;
             gcli.proc_everything().await?;
         }
         StravaActivity::fix_summary_id_in_db(pool).await?;
@@ -145,7 +145,7 @@ impl FitbitTcxSyncRequest {
         let filenames = client.sync_tcx(start_date).await?;
 
         let gcli = GarminCli::from_pool(pool)?;
-        gcli.sync_everything(false).await?;
+        gcli.sync_everything().await?;
         gcli.proc_everything().await?;
         Ok(filenames)
     }

garmin_http/src/garmin_rust_app.rs

@@ -71,7 +71,7 @@ pub async fn start_app() -> Result<(), Error> {
             {
                 if !filenames.is_empty() || !input_files.is_empty() || !dates.is_empty() {
                     info!("processed {filenames:?} and {input_files:?}");
-                    for line in cli.sync_everything(false).await.unwrap_or(Vec::new()) {
+                    for line in cli.sync_everything().await.unwrap_or(Vec::new()) {
                         info!("{line}");
                     }
                     if let Ok(client) = FitbitClient::with_auth(cli.config.clone()).await {

garmin_http/src/garmin_rust_routes.rs

@@ -313,7 +313,7 @@ async fn garmin_upload_body(
     let gcli = GarminCli::from_pool(&state.db)?;
     let filenames = vec![filename];
     let datetimes = gcli.process_filenames(&filenames).await?;
-    gcli.sync_everything(false).await?;
+    gcli.sync_everything().await?;
     gcli.proc_everything().await?;
 
     let query = FilterRequest {
@@ -332,11 +332,16 @@ async fn garmin_upload_body(
 async fn save_file(file_path: &str, field: Part) -> Result<u64, anyhow::Error> {
     let mut file = File::create(file_path).await?;
     let mut stream = field.stream();
+    let mut buf_size = 0usize;
 
     while let Some(chunk) = stream.next().await {
-        file.write_all(chunk?.chunk()).await?;
+        let chunk = chunk?;
+        let chunk = chunk.chunk();
+        buf_size += chunk.len();
+        file.write_all(chunk).await?;
     }
     let file_size = file.metadata().await?.len();
+    debug_assert!(buf_size as u64 == file_size);
     Ok(file_size)
 }
 
@@ -360,10 +365,7 @@ pub async fn garmin_sync(
     #[data] state: AppState,
 ) -> WarpResult<GarminSyncResponse> {
     let gcli = GarminCli::from_pool(&state.db).map_err(Into::<Error>::into)?;
-    let mut body = gcli
-        .sync_everything(false)
-        .await
-        .map_err(Into::<Error>::into)?;
+    let mut body = gcli.sync_everything().await.map_err(Into::<Error>::into)?;
     body.extend_from_slice(&gcli.proc_everything().await.map_err(Into::<Error>::into)?);
     let body = body.join("\n").into();
     let body = table_body(body).into();

garmin_http/src/lib.rs

@@ -8,6 +8,7 @@
 #![allow(clippy::similar_names)]
 #![allow(clippy::unused_async)]
 #![allow(clippy::unsafe_derive_deserialize)]
+#![allow(clippy::ignored_unit_patterns)]
 
 pub mod errors;
 pub mod garmin_elements;

garmin_http/src/logged_user.rs

@@ -164,11 +164,11 @@ impl Session {
     pub fn get_jwt_cookie(&self, domain: &str) -> Cookie<'static> {
         let history_str = self.history.join(";");
         let token = STANDARD.encode(history_str);
-        Cookie::build("session", token)
+        Cookie::build(("session", token))
             .http_only(true)
             .path("/")
             .domain(domain.to_string())
-            .finish()
+            .build()
     }
 }
 

garmin_lib/Cargo.toml

@@ -7,8 +7,8 @@ edition = "2018"
 [dependencies]
 anyhow = "1.0"
 avro-rs = {version = "0.13", features = ["snappy"]}
-aws-config = "0.56"
-aws-sdk-s3 = "0.31"
+aws-config = "0.101"
+aws-sdk-s3 = "0.38"
 base64 = "0.21"
 bytes = "1.0"
 deadpool = "0.10"
@@ -20,7 +20,7 @@ envy = "0.4"
 fitparser = {git="https://github.com/ddboline/fitparse-rs.git", branch="time-0.3-0.5.2"}
 flate2 = "1.0"
 futures = "0.3"
-itertools = "0.11"
+itertools = "0.12"
 json = "0.12"
 lazy_static = "1.4"
 log = "0.4"
@@ -31,7 +31,7 @@ postgres-types = {version="0.2", features=["with-time-0_3", "with-uuid-1", "with
 rand = "0.8"
 rayon = "1.5"
 regex = "1.4"
-roxmltree = "0.18"
+roxmltree = "0.19"
 rustls-native-certs = {version="0.6", default_features=false}
 serde = {version="1.0", features=["derive"]}
 serde_json = "1.0"

garmin_lib/src/common/garmin_correction_lap.rs

@@ -342,7 +342,7 @@ impl GarminCorrectionLap {
     pub async fn fix_corrections_in_db(pool: &PgPool) -> Result<(), Error> {
         let query = query!(
             "
-            UPDATE garmin_corrections_laps_backup SET summary_id = (
+            UPDATE garmin_corrections_laps SET summary_id = (
                 SELECT id FROM garmin_summary a WHERE a.begin_datetime = start_time
             )
             WHERE summary_id IS NULL