Support missing ClickHouse connection

apps/labrinth/src/background_task.rs

@@ -24,7 +24,7 @@ impl BackgroundTask {
         pool: sqlx::Pool<Postgres>,
         redis_pool: RedisPool,
         search_config: search::SearchConfig,
-        clickhouse: clickhouse::Client,
+        clickhouse: Option<clickhouse::Client>,
         stripe_client: stripe::Client,
     ) {
         use BackgroundTask::*;
@@ -33,7 +33,13 @@ impl BackgroundTask {
             IndexSearch => index_search(pool, redis_pool, search_config).await,
             ReleaseScheduled => release_scheduled(pool).await,
             UpdateVersions => update_versions(pool, redis_pool).await,
-            Payouts => payouts(pool, clickhouse).await,
+            Payouts => {
+                if let Some(client) = clickhouse {
+                    payouts(pool, Some(client)).await
+                } else {
+                    warn!("Cannot run payouts: ClickHouse client unavailable");
+                }
+            }
             IndexBilling => {
                 crate::routes::internal::billing::index_billing(
                     stripe_client,
@@ -121,14 +127,18 @@ pub async fn update_versions(
 
 pub async fn payouts(
     pool: sqlx::Pool<Postgres>,
-    clickhouse: clickhouse::Client,
+    clickhouse: Option<clickhouse::Client>,
 ) {
-    info!("Started running payouts");
-    let result = process_payout(&pool, &clickhouse).await;
-    if let Err(e) = result {
-        warn!("Payouts run failed: {:?}", e);
+    if let Some(client) = clickhouse {
+        info!("Started running payouts");
+        let result = process_payout(&pool, Some(&client)).await;
+        if let Err(e) = result {
+            warn!("Payouts run failed: {:?}", e);
+        }
+        info!("Done running payouts");
+    } else {
+        warn!("Skipping payouts: ClickHouse client unavailable");
     }
-    info!("Done running payouts");
 }
 
 mod version_updater {

apps/labrinth/src/clickhouse/fetch.rs

@@ -1,5 +1,3 @@
-use std::sync::Arc;
-
 use crate::{models::ids::ProjectId, routes::ApiError};
 use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize};
@@ -25,7 +23,7 @@ pub async fn fetch_playtimes(
     start_date: DateTime<Utc>,
     end_date: DateTime<Utc>,
     resolution_minute: u32,
-    client: Arc<clickhouse::Client>,
+    client: &clickhouse::Client,
 ) -> Result<Vec<ReturnIntervals>, ApiError> {
     let query = client
         .query(
@@ -56,12 +54,12 @@ pub async fn fetch_views(
     start_date: DateTime<Utc>,
     end_date: DateTime<Utc>,
     resolution_minutes: u32,
-    client: Arc<clickhouse::Client>,
+    client: &clickhouse::Client,
 ) -> Result<Vec<ReturnIntervals>, ApiError> {
     let query = client
         .query(
             "
-            SELECT  
+            SELECT
                 toUnixTimestamp(toStartOfInterval(recorded, toIntervalMinute(?))) AS time,
                 project_id AS id,
                 count(1) AS total
@@ -86,12 +84,12 @@ pub async fn fetch_downloads(
     start_date: DateTime<Utc>,
     end_date: DateTime<Utc>,
     resolution_minutes: u32,
-    client: Arc<clickhouse::Client>,
+    client: &clickhouse::Client,
 ) -> Result<Vec<ReturnIntervals>, ApiError> {
     let query = client
         .query(
             "
-            SELECT  
+            SELECT
                 toUnixTimestamp(toStartOfInterval(recorded, toIntervalMinute(?))) AS time,
                 project_id as id,
                 count(1) AS total
@@ -113,7 +111,7 @@ pub async fn fetch_countries_downloads(
     projects: Vec<ProjectId>,
     start_date: DateTime<Utc>,
     end_date: DateTime<Utc>,
-    client: Arc<clickhouse::Client>,
+    client: &clickhouse::Client,
 ) -> Result<Vec<ReturnCountry>, ApiError> {
     let query = client
         .query(
@@ -140,7 +138,7 @@ pub async fn fetch_countries_views(
     projects: Vec<ProjectId>,
     start_date: DateTime<Utc>,
     end_date: DateTime<Utc>,
-    client: Arc<clickhouse::Client>,
+    client: &clickhouse::Client,
 ) -> Result<Vec<ReturnCountry>, ApiError> {
     let query = client
         .query(

apps/labrinth/src/lib.rs

@@ -44,7 +44,7 @@ pub struct Pepper {
 pub struct LabrinthConfig {
     pub pool: sqlx::Pool<Postgres>,
     pub redis_pool: RedisPool,
-    pub clickhouse: Client,
+    pub clickhouse: Option<Client>,
     pub file_host: Arc<dyn file_hosting::FileHost + Send + Sync>,
     pub maxmind: Arc<queue::maxmind::MaxMindIndexer>,
     pub scheduler: Arc<scheduler::Scheduler>,
@@ -64,7 +64,7 @@ pub fn app_setup(
     pool: sqlx::Pool<Postgres>,
     redis_pool: RedisPool,
     search_config: search::SearchConfig,
-    clickhouse: &mut Client,
+    clickhouse: Option<Client>,
     file_host: Arc<dyn file_hosting::FileHost + Send + Sync>,
     maxmind: Arc<queue::maxmind::MaxMindIndexer>,
     stripe_client: stripe::Client,
@@ -142,15 +142,24 @@ pub fn app_setup(
             }
         });
 
-        let pool_ref = pool.clone();
-        let client_ref = clickhouse.clone();
-        scheduler.run(Duration::from_secs(60 * 60 * 6), move || {
-            let pool_ref = pool_ref.clone();
-            let client_ref = client_ref.clone();
-            async move {
-                background_task::payouts(pool_ref, client_ref).await;
+        match clickhouse.clone() {
+            Some(client) => {
+                let pool_ref = pool.clone();
+                scheduler.run(Duration::from_secs(60 * 60 * 6), move || {
+                    let pool_ref = pool_ref.clone();
+                    let client_ref = client.clone();
+                    async move {
+                        background_task::payouts(pool_ref, Some(client_ref))
+                            .await;
+                    }
+                });
             }
-        });
+            None => {
+                warn!(
+                    "Skipping scheduling payouts task: ClickHouse client unavailable"
+                );
+            }
+        }
 
         let pool_ref = pool.clone();
         let redis_ref = redis_pool.clone();
@@ -224,27 +233,35 @@ pub fn app_setup(
 
     let analytics_queue = Arc::new(AnalyticsQueue::new());
     {
-        let client_ref = clickhouse.clone();
-        let analytics_queue_ref = analytics_queue.clone();
-        let pool_ref = pool.clone();
-        let redis_ref = redis_pool.clone();
-        scheduler.run(Duration::from_secs(15), move || {
-            let client_ref = client_ref.clone();
-            let analytics_queue_ref = analytics_queue_ref.clone();
-            let pool_ref = pool_ref.clone();
-            let redis_ref = redis_ref.clone();
-
-            async move {
-                info!("Indexing analytics queue");
-                let result = analytics_queue_ref
-                    .index(client_ref, &redis_ref, &pool_ref)
-                    .await;
-                if let Err(e) = result {
-                    warn!("Indexing analytics queue failed: {:?}", e);
-                }
-                info!("Done indexing analytics queue");
+        match clickhouse.clone() {
+            Some(client) => {
+                let analytics_queue_ref = analytics_queue.clone();
+                let pool_ref = pool.clone();
+                let redis_ref = redis_pool.clone();
+                scheduler.run(Duration::from_secs(15), move || {
+                    let client_ref = client.clone();
+                    let analytics_queue_ref = analytics_queue_ref.clone();
+                    let pool_ref = pool_ref.clone();
+                    let redis_ref = redis_ref.clone();
+
+                    async move {
+                        info!("Indexing analytics queue");
+                        let result = analytics_queue_ref
+                            .index(Some(client_ref), &redis_ref, &pool_ref)
+                            .await;
+                        if let Err(e) = result {
+                            warn!("Indexing analytics queue failed: {:?}", e);
+                        }
+                        info!("Done indexing analytics queue");
+                    }
+                });
             }
-        });
+            None => {
+                warn!(
+                    "Skipping scheduling analytics index: ClickHouse client unavailable"
+                );
+            }
+        }
     }
 
     let ip_salt = Pepper {
@@ -285,7 +302,7 @@ pub fn app_setup(
     LabrinthConfig {
         pool,
         redis_pool,
-        clickhouse: clickhouse.clone(),
+        clickhouse,
         file_host,
         maxmind,
         scheduler: Arc::new(scheduler),

apps/labrinth/src/main.rs

@@ -11,7 +11,7 @@
 use labrinth::{check_env_vars, clickhouse, database, file_hosting, queue};
 use std::ffi::CStr;
 use std::sync::Arc;
-use tracing::{error, info};
+use tracing::{error, info, warn};
 use tracing_actix_web::TracingLogger;
 
 #[cfg(target_os = "linux")]
@@ -126,8 +126,19 @@
             _ => panic!("Invalid storage backend specified. Aborting startup!"),
         };
 
-    info!("Initializing clickhouse connection");
-    let mut clickhouse = clickhouse::init_client().await.unwrap();
+    info!("Initializing ClickHouse connection");
+    let clickhouse = clickhouse::init_client()
+        .await
+        .inspect_err(|err| {
+            warn!("ClickHouse connection unavailable: {err}.");
+
+            let hard_fail = dotenvy::var("CLICKHOUSE_HARD_FAIL").is_ok_and(|v| !v.is_empty());
+
+            if hard_fail {
+                panic!("Aborting because a ClickHouse connection couldn't be established, and CLICKHOUSE_HAIL_FAIL is set.")
+            }
+        })
+        .ok();
 
     let search_config = search::SearchConfig::new(None);
 
@@ -136,8 +147,14 @@
 
     if let Some(task) = args.run_background_task {
         info!("Running task {task:?} and exiting");
-        task.run(pool, redis_pool, search_config, clickhouse, stripe_client)
-            .await;
+        task.run(
+            pool,
+            redis_pool,
+            search_config,
+            clickhouse.clone(),
+            stripe_client,
+        )
+        .await;
         return Ok(());
     }
 
@@ -169,7 +186,7 @@
         pool.clone(),
         redis_pool.clone(),
         search_config.clone(),
-        &mut clickhouse,
+        clickhouse,
         file_host.clone(),
         maxmind_reader.clone(),
         stripe_client,

apps/labrinth/src/queue/analytics.rs

@@ -52,7 +52,7 @@ impl AnalyticsQueue {
 
     pub async fn index(
         &self,
-        client: clickhouse::Client,
+        client: Option<clickhouse::Client>,
         redis: &RedisPool,
         pool: &PgPool,
     ) -> Result<(), ApiError> {
@@ -65,6 +65,11 @@ impl AnalyticsQueue {
         let playtime_queue = self.playtime_queue.clone();
         self.playtime_queue.clear();
 
+        let Some(client) = client else {
+            // If ClickHouse isn't available, skip indexing silently
+            return Ok(());
+        };
+
         if !playtime_queue.is_empty() {
             let mut playtimes = client.insert("playtime")?;
 

apps/labrinth/src/queue/payouts.rs

@@ -736,8 +736,12 @@ pub async fn make_aditude_request(
 
 pub async fn process_payout(
     pool: &PgPool,
-    client: &clickhouse::Client,
+    client: Option<&clickhouse::Client>,
 ) -> Result<(), ApiError> {
+    let Some(client) = client else {
+        return Ok(());
+    };
+
     sqlx::query!(
         "
         UPDATE payouts