feat(crates/yt/update): Make the concurrency configurable

2 files changed, 20 insertions, 23 deletions

diff --git a/crates/yt/src/config/mod.rs b/crates/yt/src/config/mod.rs
index 154a109..4a9fa94 100644
--- a/crates/yt/src/config/mod.rs
+++ b/crates/yt/src/config/mod.rs
@@ -69,6 +69,14 @@ mk_config! {
         update: UpdateConfig = {
             /// How many videos to download, when checking for new ones.
             max_backlog: usize =: 20,
+
+            /// How many threads to use in the thread pool for fetching new videos.
+            pool_size: usize =: 16,
+
+            /// How many subscriptions to fetch at once.
+            ///
+            /// For example, 16 means, that we will fetch 16 subscriptions at the same time.
+            futures: usize =: 16 * 4,
         },
     }
 }
diff --git a/crates/yt/src/update/updater.rs b/crates/yt/src/update/updater.rs
index 3e3ceeb..ae9acb1 100644
--- a/crates/yt/src/update/updater.rs
+++ b/crates/yt/src/update/updater.rs
@@ -14,36 +14,34 @@ use std::{
 };
 
 use anyhow::{Context, Result};
-use blake3::Hash;
 use futures::{StreamExt, future::join_all, stream};
 use log::{Level, debug, error, log_enabled};
 use serde_json::json;
 use tokio_util::task::LocalPoolHandle;
 use yt_dlp::{
-    info_json::InfoJson, json_cast, json_get, options::YoutubeDLOptions, process_ie_result,
+    info_json::InfoJson, json_cast, options::YoutubeDLOptions, process_ie_result,
     python_error::PythonError,
 };
 
 use crate::{
     ansi_escape_codes::{clear_whole_line, move_to_col},
     app::App,
-    storage::subscriptions::Subscription,
+    storage::db::{extractor_hash::ExtractorHash, subscription::Subscription},
 };
 
 use super::process_subscription;
 
 pub(super) struct Updater {
     max_backlog: usize,
-    hashes: Vec<Hash>,
+    hashes: Vec<ExtractorHash>,
     pool: LocalPoolHandle,
 }
 
 static REACHED_NUMBER: AtomicUsize = const { AtomicUsize::new(1) };
 
 impl Updater {
-    pub(super) fn new(max_backlog: usize, hashes: Vec<ExtractorHash>) -> Self {
-        // TODO(@bpeetz): The number should not be hardcoded. <2025-06-14>
-        let pool = LocalPoolHandle::new(16);
+    pub(super) fn new(max_backlog: usize, max_threads: usize, hashes: Vec<ExtractorHash>) -> Self {
+        let pool = LocalPoolHandle::new(max_threads);
 
         Self {
             max_backlog,
@@ -52,22 +50,12 @@ impl Updater {
         }
     }
 
-    pub(super) async fn update(
-        self,
-        app: &App,
-        subscriptions: Vec<Subscription>,
-        total_number: Option<usize>,
-        current_progress: Option<usize>,
-    ) -> Result<()> {
-        let total_number = total_number.unwrap_or(subscriptions.len());
-
-        if let Some(current_progress) = current_progress {
-            REACHED_NUMBER.store(current_progress, Ordering::Relaxed);
-        }
+    pub(super) async fn update(self, app: &App, subscriptions: Vec<Subscription>) -> Result<()> {
+        let total_number = subscriptions.len();
 
         let mut stream = stream::iter(subscriptions)
             .map(|sub| self.get_new_entries(sub, total_number))
-            .buffer_unordered(16 * 4);
+            .buffer_unordered(app.config.update.futures);
 
         while let Some(output) = stream.next().await {
             let mut entries = output?;
@@ -86,6 +74,7 @@ impl Updater {
         Ok(())
     }
 
+    #[allow(clippy::too_many_lines)]
     async fn get_new_entries(
         &self,
         sub: Subscription,
@@ -102,9 +91,9 @@ impl Updater {
                 "extractor_args",
                 json! {{"youtubetab": {"approximate_date": [""]}}},
             )
-            // TODO: This also removes unlisted and other stuff. Find a good way to remove the
-            // members-only videos from the feed. <2025-04-17>
-            .set("match-filter", "availability=public")
+            // // TODO: This also removes unlisted and other stuff. Find a good way to remove the
+            // // members-only videos from the feed. <2025-04-17>
+            // .set("match-filter", "availability=public")
             .build()?;
 
         self.pool