bugfix: improved error handling

Author: akorchyn

server/src/contract_pull.rs

@@ -3,6 +3,7 @@ use std::{
     time::Duration,
 };
 
+use anyhow::Context;
 use chrono::DateTime;
 use rocket::fairing::AdHoc;
 use rocket_db_pools::Database;
@@ -12,7 +13,6 @@ use sqlx::{Postgres, Transaction};
 use crate::db::DB;
 
 async fn fetch_and_store_users(
-    telegram: &Arc<TelegramSubscriber>,
     near_client: &NearClient,
     tx: &mut Transaction<'static, Postgres>,
 ) -> anyhow::Result<()> {
@@ -23,61 +23,61 @@ async fn fetch_and_store_users(
         .into_iter()
         .map(|e| e.time_string(timestamp as u64))
         .collect();
-    let users = near_client.users(periods).await?;
+    let users = near_client
+        .users(periods)
+        .await
+        .context("Failed to fetch users")?;
 
     for user in users {
-        let user_id = match DB::upsert_user(tx, user.id, &user.name, user.percentage_bonus).await {
-            Ok(id) => id,
-            Err(e) => {
-                crate::error(
-                    telegram,
-                    &format!("Failed to upsert user ({}): {:#?}", user.name, e),
-                );
-                continue;
-            }
-        };
+        let user_id = DB::upsert_user(tx, user.id, &user.name, user.percentage_bonus)
+            .await
+            .with_context(|| format!("Failed to upsert user with id: {}", user.id))?;
         for (period, data) in user.period_data {
-            if let Err(e) = DB::upsert_user_period_data(tx, period, &data, user_id).await {
-                crate::error(
-                    telegram,
-                    &format!(
-                        "Failed to upsert user ({}) period data: {:#?}",
-                        user.name, e
-                    ),
-                );
-            }
+            DB::upsert_user_period_data(tx, period, &data, user_id)
+                .await
+                .with_context(|| {
+                    format!("Failed to upsert period data for user id: {}", user_id)
+                })?;
         }
         for (streak_id, streak_data) in user.streaks {
-            if let Err(e) =
-                DB::upsert_streak_user_data(tx, &streak_data, streak_id as i32, user_id).await
-            {
-                crate::error(
-                    telegram,
-                    &format!(
-                        "Failed to upsert user ({}) streak data: {:#?}",
-                        user.name, e
-                    ),
-                );
-            }
+            DB::upsert_streak_user_data(tx, &streak_data, streak_id as i32, user_id)
+                .await
+                .with_context(|| {
+                    format!(
+                        "Failed to upsert streak data for user id: {} and streak id: {}",
+                        user_id, streak_id
+                    )
+                })?;
         }
     }
 
     Ok(())
 }
 
 async fn fetch_and_store_prs(
-    telegram: &Arc<TelegramSubscriber>,
     near_client: &NearClient,
     tx: &mut Transaction<'static, Postgres>,
 ) -> anyhow::Result<()> {
-    let prs = near_client.prs().await?;
+    let prs = near_client
+        .prs()
+        .await
+        .context("Failed to fetch PRs from near_client")?;
+
+    DB::clear_prs(tx)
+        .await
+        .context("Failed to clear existing PRs from the database")?;
 
-    DB::clear_prs(tx).await?;
     for (pr, executed) in prs {
-        let organization_id = DB::upsert_organization(tx, &pr.organization).await?;
-        let repo_id = DB::upsert_repo(tx, organization_id, &pr.repo).await?;
-        let author_id = DB::get_user_id(tx, &pr.author).await?;
-        if let Err(e) = DB::upsert_pull_request(
+        let organization_id = DB::upsert_organization(tx, &pr.organization)
+            .await
+            .context("Failed on upserting organization")?;
+        let repo_id = DB::upsert_repo(tx, organization_id, &pr.repo)
+            .await
+            .context("Failed on upserting repo")?;
+        let author_id = DB::get_user_id(tx, &pr.author)
+            .await
+            .context("Failed on getting user id")?;
+        DB::upsert_pull_request(
             tx,
             repo_id,
             pr.number as i32,
@@ -93,69 +93,44 @@ async fn fetch_and_store_prs(
             executed,
         )
         .await
-        {
-            crate::error(
-                telegram,
-                &format!(
-                    "Failed to upsert PR ({}/{}/pull/{}): {:#?}",
-                    pr.organization, pr.repo, pr.number, e
-                ),
-            );
-        }
+        .context("Failed on upserting PR")?;
     }
 
     Ok(())
 }
 
 async fn fetch_and_store_repos(
-    telegram: &Arc<TelegramSubscriber>,
     near_client: &NearClient,
     tx: &mut Transaction<'static, Postgres>,
 ) -> anyhow::Result<()> {
     let organizations = near_client.repos().await?;
     for org in organizations {
-        let organization_id = match DB::upsert_organization(tx, &org.organization).await {
-            Ok(id) => id,
-            Err(e) => {
-                crate::error(
-                    telegram,
-                    &format!(
-                        "Failed to upsert organization ({}): {:#?}",
-                        org.organization, e
-                    ),
-                );
-                continue;
-            }
-        };
+        let organization_id = DB::upsert_organization(tx, &org.organization)
+            .await
+            .context("Failed on upserting organization")?;
         for repo in org.repos {
-            if let Err(e) = DB::upsert_repo(tx, organization_id, &repo).await {
-                crate::error(
-                    telegram,
-                    &format!(
-                        "Failed to upsert repo ({}/{}): {:#?}",
-                        org.organization, repo, e
-                    ),
-                );
-            }
+            DB::upsert_repo(tx, organization_id, &repo)
+                .await
+                .context("Failed on upserting repo")?;
         }
     }
 
     Ok(())
 }
 
 // TODO: more efficient way to fetch only updated data
-async fn fetch_and_store_all_data(
-    telegram: &Arc<TelegramSubscriber>,
-    near_client: &NearClient,
-    db: &DB,
-) -> anyhow::Result<()> {
+async fn fetch_and_store_all_data(near_client: &NearClient, db: &DB) -> anyhow::Result<()> {
     let mut tx = db.begin().await?;
 
-    fetch_and_store_users(telegram, near_client, &mut tx).await?;
-
-    fetch_and_store_repos(telegram, near_client, &mut tx).await?;
-    // It matters that we fetch users first, because we need to know their IDs
-    fetch_and_store_prs(telegram, near_client, &mut tx).await?;
+    fetch_and_store_users(near_client, &mut tx)
+        .await
+        .context("Failed to fetch and store users")?;
+    fetch_and_store_repos(near_client, &mut tx)
+        .await
+        .context("Failed to fetch and store repositories")?;
+    fetch_and_store_prs(near_client, &mut tx)
+        .await
+        .context("Failed to fetch and store pull requests")?;
 
     tx.commit().await?;
     Ok(())
@@ -180,11 +155,8 @@ pub fn stage(client: NearClient, sleep_duration: Duration, atomic_bool: Arc<Atom
                     interval.tick().await;
 
                     // Execute a query of some kind
-                    if let Err(e) = fetch_and_store_all_data(&telegram, &near_client, &db).await {
-                        crate::error(
-                            &telegram,
-                            &format!("Failed to fetch and store data: {:#?}", e),
-                        );
+                    if let Err(e) = fetch_and_store_all_data(&near_client, &db).await {
+                        crate::error(&telegram, &e.to_string());
                     }
                 }
             });

server/src/github_pull.rs

@@ -3,14 +3,15 @@ use std::{
     time::Duration,
 };
 
+use anyhow::Context;
 use octocrab::{models::pulls::PullRequest, Octocrab};
 use rocket::fairing::AdHoc;
 use rocket_db_pools::Database;
 use shared::telegram::TelegramSubscriber;
 use sqlx::{Postgres, Transaction};
 use tracing::instrument;
 
-use crate::db::DB;
+use crate::{db::DB, error};
 
 struct RepoMetadata {
     stars: u32,
@@ -100,7 +101,7 @@ async fn fetch_repos_metadata(
     github: &GithubClient,
     tx: &mut Transaction<'static, Postgres>,
 ) -> anyhow::Result<()> {
-    let repos = DB::get_repos(tx).await?;
+    let repos = DB::get_repos(tx).await.context("Failed to ger repops")?;
     for repo in repos {
         let metadata = match github.repo_metadata(&repo.organization, &repo.repo).await {
             Ok(metadata) => metadata,
@@ -115,7 +116,7 @@ async fn fetch_repos_metadata(
                 continue;
             }
         };
-        if let Err(e) = DB::update_repo_metadata(
+        DB::update_repo_metadata(
             tx,
             repo.repo_id,
             metadata.stars,
@@ -124,15 +125,7 @@ async fn fetch_repos_metadata(
             metadata.primary_language,
         )
         .await
-        {
-            crate::error(
-                telegram,
-                &format!(
-                    "Failed to update repo metadata for {}/{}: {:#?}",
-                    &repo.organization, &repo.repo, e
-                ),
-            );
-        }
+        .context("Failed to update repo metadata")?;
     }
     Ok(())
 }
@@ -143,7 +136,7 @@ async fn fetch_missing_user_organization_metadata(
     github: &GithubClient,
     tx: &mut Transaction<'static, Postgres>,
 ) -> anyhow::Result<()> {
-    let users = DB::get_users(tx).await?;
+    let users = DB::get_users(tx).await.unwrap_or_default();
     for user in users {
         if user.full_name.is_some() {
             // TODO: add user entry to sync cache
@@ -161,11 +154,13 @@ async fn fetch_missing_user_organization_metadata(
             }
         };
         if let Some(full_name) = &profile.name {
-            DB::update_user_full_name(tx, &user.login, full_name).await?;
+            DB::update_user_full_name(tx, &user.login, full_name)
+                .await
+                .context("Failed to update user full name")?;
         }
     }
 
-    let orgs = DB::get_organizations(tx).await?;
+    let orgs = DB::get_organizations(tx).await.unwrap_or_default();
     for org in orgs {
         if org.full_name.is_some() {
             continue;
@@ -185,7 +180,9 @@ async fn fetch_missing_user_organization_metadata(
             }
         };
         if let Some(full_name) = &profile.name {
-            DB::update_organization_full_name(tx, &org.login, full_name).await?;
+            DB::update_organization_full_name(tx, &org.login, full_name)
+                .await
+                .context("Failed to update organization full name")?;
         }
     }
     Ok(())
@@ -198,8 +195,12 @@ pub async fn fetch_github_data(
 ) -> anyhow::Result<()> {
     let mut tx = db.begin().await?;
 
-    fetch_repos_metadata(telegram, github, &mut tx).await?;
-    fetch_missing_user_organization_metadata(telegram, github, &mut tx).await?;
+    fetch_repos_metadata(telegram, github, &mut tx)
+        .await
+        .context("Failure on fetching and updating repos")?;
+    fetch_missing_user_organization_metadata(telegram, github, &mut tx)
+        .await
+        .context("Failed ton fetching and updating user/org metadata")?;
 
     tx.commit().await?;
 
@@ -239,10 +240,7 @@ pub fn stage(
                                 if let Err(e) =
                                     fetch_github_data(&telegram, &github_client, &db).await
                                 {
-                                    rocket::error!(
-                                        "Failed to fetch and store github data: {:#?}",
-                                        e
-                                    );
+                                    error(&telegram, &e.to_string());
                                 }
                             }
                         });