feat: Tauri commands, concurrent refresh, download manager
Refresh fetches 8 channel feeds concurrently and mirrors thumbnails to disk so the offline feed still renders. Downloads are capped at 2 and shell out to yt-dlp with an absolute binary path, since GUI apps do not inherit a login shell PATH.
This commit is contained in:
vincent
@@ -0,0 +1,527 @@
|
||||
//! Tauri command surface. Delegates to the modules; holds no parsing logic.
|
||||
|
||||
use crate::db::Db;
|
||||
use crate::downloader::{self, Progress};
|
||||
use crate::feed;
|
||||
use crate::models::{ChannelWithCount, DownloadState, FeedFilter, FeedItem, Prereqs};
|
||||
use crate::net;
|
||||
use crate::takeout;
|
||||
use crate::thumbs;
|
||||
|
||||
use futures::stream::StreamExt;
|
||||
use serde::Serialize;
|
||||
use std::collections::HashMap;
|
||||
use std::path::PathBuf;
|
||||
use std::sync::Arc;
|
||||
use tauri::{AppHandle, Emitter, Manager, State};
|
||||
use tokio::io::{AsyncBufReadExt, BufReader};
|
||||
use tokio::sync::{Mutex, Semaphore};
|
||||
|
||||
/// How many channel feeds we fetch at once. YouTube serves these fine in
|
||||
/// parallel; 8 keeps a 300-channel refresh brisk without hammering them.
|
||||
const FEED_CONCURRENCY: usize = 8;
|
||||
/// Video downloads are bandwidth-bound, so more than a couple at once just
|
||||
/// makes every one of them slower.
|
||||
const DOWNLOAD_CONCURRENCY: usize = 2;
|
||||
/// Cap thumbnail work per refresh so a first-run import doesn't stall for ages.
|
||||
const THUMB_BATCH: i64 = 600;
|
||||
|
||||
pub struct AppState {
|
||||
pub db: Arc<Mutex<Db>>,
|
||||
pub http: reqwest::Client,
|
||||
pub library: Arc<Mutex<PathBuf>>,
|
||||
pub app_data: PathBuf,
|
||||
pub children: Arc<Mutex<HashMap<String, tokio::process::Child>>>,
|
||||
pub download_slots: Arc<Semaphore>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Serialize)]
|
||||
struct RefreshProgress {
|
||||
done: usize,
|
||||
total: usize,
|
||||
channel: String,
|
||||
}
|
||||
|
||||
#[derive(Clone, Serialize)]
|
||||
struct DownloadProgressEvent {
|
||||
video_id: String,
|
||||
pct: Option<f64>,
|
||||
bytes_done: u64,
|
||||
bytes_total: Option<u64>,
|
||||
speed: Option<f64>,
|
||||
eta: Option<u64>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Serialize)]
|
||||
struct DownloadStateEvent {
|
||||
video_id: String,
|
||||
state: DownloadState,
|
||||
error: Option<String>,
|
||||
path: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
pub struct RefreshSummary {
|
||||
pub channels: usize,
|
||||
pub new_videos: usize,
|
||||
pub failures: Vec<String>,
|
||||
}
|
||||
|
||||
fn bin(name: &str) -> String {
|
||||
// GUI apps launched from Finder don't inherit a login shell PATH, so
|
||||
// Homebrew's bin dir is invisible to them. Prefer an absolute path when we
|
||||
// can find one, and fall back to the bare name for PATH resolution.
|
||||
for prefix in ["/opt/homebrew/bin", "/usr/local/bin", "/usr/bin"] {
|
||||
let candidate = format!("{prefix}/{name}");
|
||||
if std::path::Path::new(&candidate).exists() {
|
||||
return candidate;
|
||||
}
|
||||
}
|
||||
name.to_string()
|
||||
}
|
||||
|
||||
fn version_of(name: &str) -> Option<String> {
|
||||
std::process::Command::new(bin(name))
|
||||
.arg("--version")
|
||||
.output()
|
||||
.ok()
|
||||
.filter(|o| o.status.success())
|
||||
.map(|o| {
|
||||
String::from_utf8_lossy(&o.stdout)
|
||||
.lines()
|
||||
.next()
|
||||
.unwrap_or("")
|
||||
.trim()
|
||||
.to_string()
|
||||
})
|
||||
.filter(|s| !s.is_empty())
|
||||
}
|
||||
|
||||
#[tauri::command]
|
||||
pub async fn check_prereqs(state: State<'_, AppState>) -> Result<Prereqs, String> {
|
||||
let library = state.library.lock().await.clone();
|
||||
Ok(Prereqs {
|
||||
yt_dlp: version_of("yt-dlp"),
|
||||
ffmpeg: version_of("ffmpeg").map(|v| {
|
||||
// ffmpeg's first line is long; keep the useful head of it.
|
||||
v.split_whitespace().take(3).collect::<Vec<_>>().join(" ")
|
||||
}),
|
||||
library_path: library.to_string_lossy().to_string(),
|
||||
})
|
||||
}
|
||||
|
||||
#[tauri::command]
|
||||
pub async fn import_takeout_csv(
|
||||
path: String,
|
||||
state: State<'_, AppState>,
|
||||
) -> Result<usize, String> {
|
||||
let raw = tokio::fs::read(&path)
|
||||
.await
|
||||
.map_err(|e| format!("Cannot read {path}: {e}"))?;
|
||||
// Takeout exports are UTF-8, sometimes with a BOM.
|
||||
let text = String::from_utf8_lossy(&raw);
|
||||
let text = text.strip_prefix('\u{feff}').unwrap_or(&text);
|
||||
|
||||
let channels = takeout::parse_csv(text)?;
|
||||
if channels.is_empty() {
|
||||
return Err("No channels found in that file.".into());
|
||||
}
|
||||
let mut db = state.db.lock().await;
|
||||
db.upsert_channels(&channels)
|
||||
}
|
||||
|
||||
#[tauri::command]
|
||||
pub async fn list_channels(state: State<'_, AppState>) -> Result<Vec<ChannelWithCount>, String> {
|
||||
state.db.lock().await.list_channels()
|
||||
}
|
||||
|
||||
#[tauri::command]
|
||||
pub async fn list_feed(
|
||||
filter: FeedFilter,
|
||||
state: State<'_, AppState>,
|
||||
) -> Result<Vec<FeedItem>, String> {
|
||||
state.db.lock().await.list_feed(&filter)
|
||||
}
|
||||
|
||||
#[tauri::command]
|
||||
pub async fn get_connectivity(state: State<'_, AppState>) -> Result<bool, String> {
|
||||
Ok(net::is_online(&state.http).await)
|
||||
}
|
||||
|
||||
#[tauri::command]
|
||||
pub async fn refresh_feeds(
|
||||
app: AppHandle,
|
||||
state: State<'_, AppState>,
|
||||
) -> Result<RefreshSummary, String> {
|
||||
let channel_ids = state.db.lock().await.channel_ids()?;
|
||||
if channel_ids.is_empty() {
|
||||
return Err("No subscriptions imported yet. Import your Takeout CSV first.".into());
|
||||
}
|
||||
|
||||
let total = channel_ids.len();
|
||||
let http = state.http.clone();
|
||||
|
||||
// Fetch with bounded concurrency. One channel failing must not abort the run.
|
||||
let results = futures::stream::iter(channel_ids.into_iter().map(|cid| {
|
||||
let http = http.clone();
|
||||
async move {
|
||||
let res = feed::fetch_channel(&http, &cid).await;
|
||||
(cid, res)
|
||||
}
|
||||
}))
|
||||
.buffer_unordered(FEED_CONCURRENCY)
|
||||
.collect::<Vec<_>>()
|
||||
.await;
|
||||
|
||||
let mut new_videos = 0usize;
|
||||
let mut failures = Vec::new();
|
||||
let mut done = 0usize;
|
||||
|
||||
for (cid, res) in results {
|
||||
done += 1;
|
||||
match res {
|
||||
Ok(videos) => {
|
||||
if !videos.is_empty() {
|
||||
let mut db = state.db.lock().await;
|
||||
new_videos += db.upsert_videos(&videos)?;
|
||||
}
|
||||
}
|
||||
Err(e) => failures.push(format!("{cid}: {e}")),
|
||||
}
|
||||
let _ = app.emit(
|
||||
"refresh:progress",
|
||||
RefreshProgress {
|
||||
done,
|
||||
total,
|
||||
channel: cid,
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
cache_thumbnails(&state).await;
|
||||
|
||||
Ok(RefreshSummary {
|
||||
channels: total,
|
||||
new_videos,
|
||||
failures,
|
||||
})
|
||||
}
|
||||
|
||||
/// Mirrors thumbnails to disk so the feed still renders with no network.
|
||||
async fn cache_thumbnails(state: &State<'_, AppState>) {
|
||||
let pending = match state.db.lock().await.videos_missing_thumbs(THUMB_BATCH) {
|
||||
Ok(p) => p,
|
||||
Err(_) => return,
|
||||
};
|
||||
if pending.is_empty() {
|
||||
return;
|
||||
}
|
||||
|
||||
let dir = thumbs::cache_dir(&state.app_data);
|
||||
let http = state.http.clone();
|
||||
|
||||
let cached = futures::stream::iter(pending.into_iter().map(|(id, url)| {
|
||||
let http = http.clone();
|
||||
let dir = dir.clone();
|
||||
async move {
|
||||
let res = thumbs::cache_one(&http, &id, &url, &dir).await;
|
||||
(id, res)
|
||||
}
|
||||
}))
|
||||
.buffer_unordered(FEED_CONCURRENCY)
|
||||
.collect::<Vec<_>>()
|
||||
.await;
|
||||
|
||||
let db = state.db.lock().await;
|
||||
for (id, res) in cached {
|
||||
if let Ok(path) = res {
|
||||
let _ = db.set_thumb_path(&id, &path.to_string_lossy());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[tauri::command]
|
||||
pub async fn download_video(
|
||||
video_id: String,
|
||||
app: AppHandle,
|
||||
state: State<'_, AppState>,
|
||||
) -> Result<(), String> {
|
||||
let library = state.library.lock().await.clone();
|
||||
tokio::fs::create_dir_all(&library)
|
||||
.await
|
||||
.map_err(|e| format!("Cannot create library dir: {e}"))?;
|
||||
|
||||
{
|
||||
let db = state.db.lock().await;
|
||||
db.set_download_state(&video_id, DownloadState::Queued, None)?;
|
||||
}
|
||||
let _ = app.emit(
|
||||
"download:state",
|
||||
DownloadStateEvent {
|
||||
video_id: video_id.clone(),
|
||||
state: DownloadState::Queued,
|
||||
error: None,
|
||||
path: None,
|
||||
},
|
||||
);
|
||||
|
||||
let permit = state
|
||||
.download_slots
|
||||
.clone()
|
||||
.acquire_owned()
|
||||
.await
|
||||
.map_err(|e| format!("Download queue closed: {e}"))?;
|
||||
|
||||
let out_template = library.join("%(id)s.%(ext)s").to_string_lossy().to_string();
|
||||
let args = downloader::build_args(&video_id, &out_template);
|
||||
|
||||
let mut child = tokio::process::Command::new(bin("yt-dlp"))
|
||||
.args(&args)
|
||||
.stdout(std::process::Stdio::piped())
|
||||
.stderr(std::process::Stdio::piped())
|
||||
.spawn()
|
||||
.map_err(|e| {
|
||||
format!("Could not start yt-dlp: {e}. Install it with: brew install yt-dlp ffmpeg")
|
||||
})?;
|
||||
|
||||
let stdout = child.stdout.take().ok_or("yt-dlp produced no stdout")?;
|
||||
let stderr = child.stderr.take().ok_or("yt-dlp produced no stderr")?;
|
||||
|
||||
state
|
||||
.children
|
||||
.lock()
|
||||
.await
|
||||
.insert(video_id.clone(), child);
|
||||
|
||||
{
|
||||
let db = state.db.lock().await;
|
||||
db.set_download_state(&video_id, DownloadState::Running, None)?;
|
||||
}
|
||||
let _ = app.emit(
|
||||
"download:state",
|
||||
DownloadStateEvent {
|
||||
video_id: video_id.clone(),
|
||||
state: DownloadState::Running,
|
||||
error: None,
|
||||
path: None,
|
||||
},
|
||||
);
|
||||
|
||||
// Collect stderr concurrently so a failure has a real message, and so a
|
||||
// full stderr pipe can't deadlock the child.
|
||||
let stderr_task = tokio::spawn(async move {
|
||||
let mut lines = BufReader::new(stderr).lines();
|
||||
let mut buf = Vec::new();
|
||||
while let Ok(Some(l)) = lines.next_line().await {
|
||||
buf.push(l);
|
||||
}
|
||||
buf
|
||||
});
|
||||
|
||||
let mut final_path: Option<String> = None;
|
||||
let mut reader = BufReader::new(stdout).lines();
|
||||
while let Ok(Some(line)) = reader.next_line().await {
|
||||
if let Some(p) = downloader::parse_final_path(&line) {
|
||||
final_path = Some(p);
|
||||
continue;
|
||||
}
|
||||
if let Some(Progress {
|
||||
downloaded,
|
||||
total,
|
||||
speed,
|
||||
eta,
|
||||
}) = downloader::parse_progress_line(&line)
|
||||
{
|
||||
let pct = Progress {
|
||||
downloaded,
|
||||
total,
|
||||
speed,
|
||||
eta,
|
||||
}
|
||||
.pct();
|
||||
{
|
||||
let db = state.db.lock().await;
|
||||
let _ = db.set_download_progress(&video_id, downloaded, total, pct);
|
||||
}
|
||||
let _ = app.emit(
|
||||
"download:progress",
|
||||
DownloadProgressEvent {
|
||||
video_id: video_id.clone(),
|
||||
pct,
|
||||
bytes_done: downloaded,
|
||||
bytes_total: total,
|
||||
speed,
|
||||
eta,
|
||||
},
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
let mut child = state
|
||||
.children
|
||||
.lock()
|
||||
.await
|
||||
.remove(&video_id)
|
||||
.ok_or_else(|| "Download was cancelled.".to_string())?;
|
||||
|
||||
let status = child
|
||||
.wait()
|
||||
.await
|
||||
.map_err(|e| format!("yt-dlp did not exit cleanly: {e}"))?;
|
||||
let stderr_lines = stderr_task.await.unwrap_or_default();
|
||||
drop(permit);
|
||||
|
||||
if status.success() {
|
||||
let path = final_path.unwrap_or_else(|| {
|
||||
library
|
||||
.join(format!("{video_id}.mp4"))
|
||||
.to_string_lossy()
|
||||
.to_string()
|
||||
});
|
||||
let db = state.db.lock().await;
|
||||
db.set_download_state(&video_id, DownloadState::Done, None)?;
|
||||
db.set_download_path(&video_id, &path)?;
|
||||
drop(db);
|
||||
let _ = app.emit(
|
||||
"download:state",
|
||||
DownloadStateEvent {
|
||||
video_id,
|
||||
state: DownloadState::Done,
|
||||
error: None,
|
||||
path: Some(path),
|
||||
},
|
||||
);
|
||||
Ok(())
|
||||
} else {
|
||||
let msg = stderr_lines
|
||||
.iter()
|
||||
.rev()
|
||||
.find(|l| l.contains("ERROR"))
|
||||
.cloned()
|
||||
.unwrap_or_else(|| format!("yt-dlp exited with {status}"));
|
||||
let db = state.db.lock().await;
|
||||
db.set_download_state(&video_id, DownloadState::Failed, Some(&msg))?;
|
||||
drop(db);
|
||||
let _ = app.emit(
|
||||
"download:state",
|
||||
DownloadStateEvent {
|
||||
video_id,
|
||||
state: DownloadState::Failed,
|
||||
error: Some(msg.clone()),
|
||||
path: None,
|
||||
},
|
||||
);
|
||||
Err(msg)
|
||||
}
|
||||
}
|
||||
|
||||
#[tauri::command]
|
||||
pub async fn cancel_download(
|
||||
video_id: String,
|
||||
app: AppHandle,
|
||||
state: State<'_, AppState>,
|
||||
) -> Result<(), String> {
|
||||
if let Some(mut child) = state.children.lock().await.remove(&video_id) {
|
||||
let _ = child.kill().await;
|
||||
}
|
||||
{
|
||||
let db = state.db.lock().await;
|
||||
db.set_download_state(&video_id, DownloadState::Cancelled, None)?;
|
||||
}
|
||||
cleanup_partials(state.library.lock().await.clone(), &video_id).await;
|
||||
let _ = app.emit(
|
||||
"download:state",
|
||||
DownloadStateEvent {
|
||||
video_id,
|
||||
state: DownloadState::Cancelled,
|
||||
error: None,
|
||||
path: None,
|
||||
},
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// yt-dlp leaves `.part`, `.ytdl` and format-specific fragments behind when
|
||||
/// killed; without this the library slowly fills with dead bytes.
|
||||
async fn cleanup_partials(library: PathBuf, video_id: &str) {
|
||||
let Ok(mut entries) = tokio::fs::read_dir(&library).await else {
|
||||
return;
|
||||
};
|
||||
while let Ok(Some(entry)) = entries.next_entry().await {
|
||||
let name = entry.file_name().to_string_lossy().to_string();
|
||||
if name.starts_with(video_id) && (name.contains(".part") || name.ends_with(".ytdl")) {
|
||||
let _ = tokio::fs::remove_file(entry.path()).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[tauri::command]
|
||||
pub async fn delete_download(
|
||||
video_id: String,
|
||||
state: State<'_, AppState>,
|
||||
) -> Result<(), String> {
|
||||
let path = state.db.lock().await.get_download_path(&video_id)?;
|
||||
if let Some(p) = path {
|
||||
let _ = tokio::fs::remove_file(&p).await;
|
||||
}
|
||||
cleanup_partials(state.library.lock().await.clone(), &video_id).await;
|
||||
state.db.lock().await.clear_download(&video_id)
|
||||
}
|
||||
|
||||
#[tauri::command]
|
||||
pub async fn set_library_path(
|
||||
path: String,
|
||||
state: State<'_, AppState>,
|
||||
) -> Result<String, String> {
|
||||
let p = PathBuf::from(&path);
|
||||
tokio::fs::create_dir_all(&p)
|
||||
.await
|
||||
.map_err(|e| format!("Cannot use that folder: {e}"))?;
|
||||
*state.library.lock().await = p.clone();
|
||||
Ok(p.to_string_lossy().to_string())
|
||||
}
|
||||
|
||||
#[tauri::command]
|
||||
pub async fn open_external(app: AppHandle, url: String) -> Result<(), String> {
|
||||
use tauri_plugin_opener::OpenerExt;
|
||||
app.opener()
|
||||
.open_url(url, None::<&str>)
|
||||
.map_err(|e| format!("Could not open link: {e}"))
|
||||
}
|
||||
|
||||
pub fn default_library() -> PathBuf {
|
||||
dirs_home()
|
||||
.map(|h| h.join("Movies").join("FlightTube"))
|
||||
.unwrap_or_else(|| PathBuf::from("FlightTube"))
|
||||
}
|
||||
|
||||
fn dirs_home() -> Option<PathBuf> {
|
||||
std::env::var_os("HOME").map(PathBuf::from)
|
||||
}
|
||||
|
||||
pub fn build_state(app: &AppHandle) -> Result<AppState, String> {
|
||||
let app_data = app
|
||||
.path()
|
||||
.app_data_dir()
|
||||
.map_err(|e| format!("No app data dir: {e}"))?;
|
||||
std::fs::create_dir_all(&app_data).map_err(|e| format!("Cannot create app data dir: {e}"))?;
|
||||
|
||||
let db = Db::open(&app_data.join("flighttube.db"))?;
|
||||
let library = default_library();
|
||||
let _ = std::fs::create_dir_all(&library);
|
||||
|
||||
let http = reqwest::Client::builder()
|
||||
.user_agent("FlightTube/0.1 (+desktop)")
|
||||
.timeout(std::time::Duration::from_secs(20))
|
||||
.build()
|
||||
.map_err(|e| format!("Cannot build HTTP client: {e}"))?;
|
||||
|
||||
Ok(AppState {
|
||||
db: Arc::new(Mutex::new(db)),
|
||||
http,
|
||||
library: Arc::new(Mutex::new(library)),
|
||||
app_data,
|
||||
children: Arc::new(Mutex::new(HashMap::new())),
|
||||
download_slots: Arc::new(Semaphore::new(DOWNLOAD_CONCURRENCY)),
|
||||
})
|
||||
}
|
||||
Reference in New Issue
Block a user