feat: Takeout CSV, Atom feed, yt-dlp progress, and SQLite layers
37 unit tests covering the three parsers and the database, including that a feed refresh preserves download state and that a truncated feed response fails rather than storing partial results.
This commit is contained in:
vincent
@@ -0,0 +1,527 @@
|
||||
//! SQLite storage. The only module that speaks SQL.
|
||||
|
||||
use crate::models::{Channel, ChannelWithCount, DownloadState, FeedFilter, FeedItem, Video};
|
||||
use rusqlite::{params, Connection};
|
||||
use std::path::Path;
|
||||
|
||||
pub struct Db {
|
||||
conn: Connection,
|
||||
}
|
||||
|
||||
const SCHEMA: &str = r#"
|
||||
CREATE TABLE IF NOT EXISTS channels (
|
||||
id TEXT PRIMARY KEY,
|
||||
title TEXT NOT NULL,
|
||||
url TEXT NOT NULL,
|
||||
added_at INTEGER NOT NULL
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS videos (
|
||||
id TEXT PRIMARY KEY,
|
||||
channel_id TEXT NOT NULL,
|
||||
title TEXT NOT NULL,
|
||||
description TEXT NOT NULL DEFAULT '',
|
||||
published INTEGER NOT NULL,
|
||||
thumb_url TEXT NOT NULL DEFAULT '',
|
||||
thumb_path TEXT,
|
||||
views INTEGER NOT NULL DEFAULT 0,
|
||||
is_short INTEGER NOT NULL DEFAULT 0,
|
||||
fetched_at INTEGER NOT NULL
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_videos_published ON videos(published DESC);
|
||||
CREATE INDEX IF NOT EXISTS idx_videos_channel ON videos(channel_id);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS downloads (
|
||||
video_id TEXT PRIMARY KEY,
|
||||
state TEXT NOT NULL,
|
||||
path TEXT,
|
||||
bytes_total INTEGER,
|
||||
bytes_done INTEGER,
|
||||
pct REAL,
|
||||
speed TEXT,
|
||||
eta TEXT,
|
||||
error TEXT,
|
||||
completed_at INTEGER
|
||||
);
|
||||
"#;
|
||||
|
||||
fn now() -> i64 {
|
||||
std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.map(|d| d.as_secs() as i64)
|
||||
.unwrap_or(0)
|
||||
}
|
||||
|
||||
impl Db {
|
||||
pub fn open(path: &Path) -> Result<Db, String> {
|
||||
if let Some(parent) = path.parent() {
|
||||
std::fs::create_dir_all(parent).map_err(|e| format!("Cannot create data dir: {e}"))?;
|
||||
}
|
||||
let conn = Connection::open(path).map_err(|e| format!("Cannot open database: {e}"))?;
|
||||
Self::init(conn)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
pub fn open_in_memory() -> Result<Db, String> {
|
||||
let conn = Connection::open_in_memory().map_err(|e| e.to_string())?;
|
||||
Self::init(conn)
|
||||
}
|
||||
|
||||
fn init(conn: Connection) -> Result<Db, String> {
|
||||
conn.execute_batch(SCHEMA)
|
||||
.map_err(|e| format!("Cannot create schema: {e}"))?;
|
||||
Ok(Db { conn })
|
||||
}
|
||||
|
||||
pub fn upsert_channels(&mut self, channels: &[Channel]) -> Result<usize, String> {
|
||||
let tx = self.conn.transaction().map_err(|e| e.to_string())?;
|
||||
{
|
||||
let mut stmt = tx
|
||||
.prepare(
|
||||
"INSERT INTO channels (id, title, url, added_at) VALUES (?1, ?2, ?3, ?4)
|
||||
ON CONFLICT(id) DO UPDATE SET title=excluded.title, url=excluded.url",
|
||||
)
|
||||
.map_err(|e| e.to_string())?;
|
||||
let ts = now();
|
||||
for c in channels {
|
||||
stmt.execute(params![c.id, c.title, c.url, ts])
|
||||
.map_err(|e| e.to_string())?;
|
||||
}
|
||||
}
|
||||
tx.commit().map_err(|e| e.to_string())?;
|
||||
Ok(channels.len())
|
||||
}
|
||||
|
||||
pub fn list_channels(&self) -> Result<Vec<ChannelWithCount>, String> {
|
||||
let mut stmt = self
|
||||
.conn
|
||||
.prepare(
|
||||
"SELECT c.id, c.title, c.url,
|
||||
(SELECT COUNT(*) FROM videos v WHERE v.channel_id = c.id),
|
||||
(SELECT COUNT(*) FROM videos v
|
||||
JOIN downloads d ON d.video_id = v.id
|
||||
WHERE v.channel_id = c.id AND d.state = 'done')
|
||||
FROM channels c
|
||||
ORDER BY c.title COLLATE NOCASE ASC",
|
||||
)
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
let rows = stmt
|
||||
.query_map([], |r| {
|
||||
Ok(ChannelWithCount {
|
||||
id: r.get(0)?,
|
||||
title: r.get(1)?,
|
||||
url: r.get(2)?,
|
||||
video_count: r.get(3)?,
|
||||
downloaded_count: r.get(4)?,
|
||||
})
|
||||
})
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
rows.collect::<Result<Vec<_>, _>>()
|
||||
.map_err(|e| e.to_string())
|
||||
}
|
||||
|
||||
pub fn channel_ids(&self) -> Result<Vec<String>, String> {
|
||||
let mut stmt = self
|
||||
.conn
|
||||
.prepare("SELECT id FROM channels")
|
||||
.map_err(|e| e.to_string())?;
|
||||
let rows = stmt
|
||||
.query_map([], |r| r.get::<_, String>(0))
|
||||
.map_err(|e| e.to_string())?;
|
||||
rows.collect::<Result<Vec<_>, _>>()
|
||||
.map_err(|e| e.to_string())
|
||||
}
|
||||
|
||||
/// Upserts video metadata only. Deliberately never touches the `downloads`
|
||||
/// table, so a refresh cannot lose the record of what has been downloaded.
|
||||
/// `thumb_path` is preserved via COALESCE for the same reason.
|
||||
pub fn upsert_videos(&mut self, videos: &[Video]) -> Result<usize, String> {
|
||||
let tx = self.conn.transaction().map_err(|e| e.to_string())?;
|
||||
{
|
||||
let mut stmt = tx
|
||||
.prepare(
|
||||
"INSERT INTO videos
|
||||
(id, channel_id, title, description, published, thumb_url, views, is_short, fetched_at)
|
||||
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)
|
||||
ON CONFLICT(id) DO UPDATE SET
|
||||
channel_id = excluded.channel_id,
|
||||
title = excluded.title,
|
||||
description = excluded.description,
|
||||
published = excluded.published,
|
||||
thumb_url = excluded.thumb_url,
|
||||
views = excluded.views,
|
||||
is_short = excluded.is_short,
|
||||
fetched_at = excluded.fetched_at",
|
||||
)
|
||||
.map_err(|e| e.to_string())?;
|
||||
let ts = now();
|
||||
for v in videos {
|
||||
stmt.execute(params![
|
||||
v.id,
|
||||
v.channel_id,
|
||||
v.title,
|
||||
v.description,
|
||||
v.published,
|
||||
v.thumb_url,
|
||||
v.views,
|
||||
v.is_short as i64,
|
||||
ts
|
||||
])
|
||||
.map_err(|e| e.to_string())?;
|
||||
}
|
||||
}
|
||||
tx.commit().map_err(|e| e.to_string())?;
|
||||
Ok(videos.len())
|
||||
}
|
||||
|
||||
pub fn set_thumb_path(&self, video_id: &str, path: &str) -> Result<(), String> {
|
||||
self.conn
|
||||
.execute(
|
||||
"UPDATE videos SET thumb_path = ?2 WHERE id = ?1",
|
||||
params![video_id, path],
|
||||
)
|
||||
.map_err(|e| e.to_string())?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Videos that still need a thumbnail cached locally.
|
||||
pub fn videos_missing_thumbs(&self, limit: i64) -> Result<Vec<(String, String)>, String> {
|
||||
let mut stmt = self
|
||||
.conn
|
||||
.prepare(
|
||||
"SELECT id, thumb_url FROM videos
|
||||
WHERE thumb_path IS NULL AND thumb_url <> ''
|
||||
ORDER BY published DESC LIMIT ?1",
|
||||
)
|
||||
.map_err(|e| e.to_string())?;
|
||||
let rows = stmt
|
||||
.query_map(params![limit], |r| Ok((r.get(0)?, r.get(1)?)))
|
||||
.map_err(|e| e.to_string())?;
|
||||
rows.collect::<Result<Vec<_>, _>>()
|
||||
.map_err(|e| e.to_string())
|
||||
}
|
||||
|
||||
pub fn list_feed(&self, f: &FeedFilter) -> Result<Vec<FeedItem>, String> {
|
||||
let mut sql = String::from(
|
||||
"SELECT v.id, v.channel_id, COALESCE(c.title, ''), v.title, v.description,
|
||||
v.published, v.thumb_url, v.thumb_path, v.views, v.is_short,
|
||||
d.state, d.path, d.pct, d.error
|
||||
FROM videos v
|
||||
LEFT JOIN channels c ON c.id = v.channel_id
|
||||
LEFT JOIN downloads d ON d.video_id = v.id
|
||||
WHERE 1=1",
|
||||
);
|
||||
let mut args: Vec<Box<dyn rusqlite::ToSql>> = Vec::new();
|
||||
|
||||
if let Some(cid) = f.channel_id.as_ref().filter(|s| !s.is_empty()) {
|
||||
sql.push_str(" AND v.channel_id = ?");
|
||||
args.push(Box::new(cid.clone()));
|
||||
}
|
||||
if let Some(q) = f.search.as_ref().filter(|s| !s.trim().is_empty()) {
|
||||
sql.push_str(" AND (v.title LIKE ? COLLATE NOCASE OR c.title LIKE ? COLLATE NOCASE)");
|
||||
let pat = format!("%{}%", q.trim());
|
||||
args.push(Box::new(pat.clone()));
|
||||
args.push(Box::new(pat));
|
||||
}
|
||||
if f.downloaded_only {
|
||||
sql.push_str(" AND d.state = 'done'");
|
||||
}
|
||||
if f.hide_shorts {
|
||||
sql.push_str(" AND v.is_short = 0");
|
||||
}
|
||||
sql.push_str(" ORDER BY v.published DESC");
|
||||
if let Some(l) = f.limit {
|
||||
sql.push_str(" LIMIT ?");
|
||||
args.push(Box::new(l));
|
||||
}
|
||||
|
||||
let mut stmt = self.conn.prepare(&sql).map_err(|e| e.to_string())?;
|
||||
let refs: Vec<&dyn rusqlite::ToSql> = args.iter().map(|b| b.as_ref()).collect();
|
||||
|
||||
let rows = stmt
|
||||
.query_map(refs.as_slice(), |r| {
|
||||
let state: Option<String> = r.get(10)?;
|
||||
Ok(FeedItem {
|
||||
id: r.get(0)?,
|
||||
channel_id: r.get(1)?,
|
||||
channel_title: r.get(2)?,
|
||||
title: r.get(3)?,
|
||||
description: r.get(4)?,
|
||||
published: r.get(5)?,
|
||||
thumb_url: r.get(6)?,
|
||||
thumb_path: r.get(7)?,
|
||||
views: r.get(8)?,
|
||||
is_short: r.get::<_, i64>(9)? != 0,
|
||||
state: state.as_deref().and_then(DownloadState::from_str),
|
||||
path: r.get(11)?,
|
||||
pct: r.get(12)?,
|
||||
error: r.get(13)?,
|
||||
})
|
||||
})
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
rows.collect::<Result<Vec<_>, _>>()
|
||||
.map_err(|e| e.to_string())
|
||||
}
|
||||
|
||||
pub fn set_download_state(
|
||||
&self,
|
||||
video_id: &str,
|
||||
state: DownloadState,
|
||||
error: Option<&str>,
|
||||
) -> Result<(), String> {
|
||||
let completed = matches!(state, DownloadState::Done).then(now);
|
||||
self.conn
|
||||
.execute(
|
||||
"INSERT INTO downloads (video_id, state, error, completed_at)
|
||||
VALUES (?1, ?2, ?3, ?4)
|
||||
ON CONFLICT(video_id) DO UPDATE SET
|
||||
state = excluded.state,
|
||||
error = excluded.error,
|
||||
completed_at = COALESCE(excluded.completed_at, downloads.completed_at)",
|
||||
params![video_id, state.as_str(), error, completed],
|
||||
)
|
||||
.map_err(|e| e.to_string())?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn set_download_progress(
|
||||
&self,
|
||||
video_id: &str,
|
||||
done: u64,
|
||||
total: Option<u64>,
|
||||
pct: Option<f64>,
|
||||
) -> Result<(), String> {
|
||||
self.conn
|
||||
.execute(
|
||||
"UPDATE downloads SET bytes_done = ?2, bytes_total = ?3, pct = ?4
|
||||
WHERE video_id = ?1",
|
||||
params![
|
||||
video_id,
|
||||
done as i64,
|
||||
total.map(|t| t as i64),
|
||||
pct
|
||||
],
|
||||
)
|
||||
.map_err(|e| e.to_string())?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn set_download_path(&self, video_id: &str, path: &str) -> Result<(), String> {
|
||||
self.conn
|
||||
.execute(
|
||||
"UPDATE downloads SET path = ?2 WHERE video_id = ?1",
|
||||
params![video_id, path],
|
||||
)
|
||||
.map_err(|e| e.to_string())?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn get_download_path(&self, video_id: &str) -> Result<Option<String>, String> {
|
||||
self.conn
|
||||
.query_row(
|
||||
"SELECT path FROM downloads WHERE video_id = ?1",
|
||||
params![video_id],
|
||||
|r| r.get::<_, Option<String>>(0),
|
||||
)
|
||||
.or_else(|e| match e {
|
||||
rusqlite::Error::QueryReturnedNoRows => Ok(None),
|
||||
other => Err(other),
|
||||
})
|
||||
.map_err(|e| e.to_string())
|
||||
}
|
||||
|
||||
pub fn clear_download(&self, video_id: &str) -> Result<(), String> {
|
||||
self.conn
|
||||
.execute("DELETE FROM downloads WHERE video_id = ?1", params![video_id])
|
||||
.map_err(|e| e.to_string())?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn video_title(&self, video_id: &str) -> Result<String, String> {
|
||||
self.conn
|
||||
.query_row(
|
||||
"SELECT title FROM videos WHERE id = ?1",
|
||||
params![video_id],
|
||||
|r| r.get(0),
|
||||
)
|
||||
.map_err(|e| e.to_string())
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn vid(id: &str, chan: &str, published: i64, short: bool) -> Video {
|
||||
Video {
|
||||
id: id.into(),
|
||||
channel_id: chan.into(),
|
||||
title: format!("Title {id}"),
|
||||
description: "d".into(),
|
||||
published,
|
||||
thumb_url: format!("https://i.ytimg.com/vi/{id}/hq.jpg"),
|
||||
views: 10,
|
||||
is_short: short,
|
||||
}
|
||||
}
|
||||
|
||||
fn seeded() -> Db {
|
||||
let mut db = Db::open_in_memory().unwrap();
|
||||
db.upsert_channels(&[
|
||||
Channel { id: "UC1".into(), title: "Alpha".into(), url: "u1".into() },
|
||||
Channel { id: "UC2".into(), title: "Beta".into(), url: "u2".into() },
|
||||
])
|
||||
.unwrap();
|
||||
db.upsert_videos(&[
|
||||
vid("a", "UC1", 100, false),
|
||||
vid("b", "UC2", 300, false),
|
||||
vid("c", "UC1", 200, true),
|
||||
])
|
||||
.unwrap();
|
||||
db
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn upserting_same_channel_twice_yields_one_row() {
|
||||
let mut db = Db::open_in_memory().unwrap();
|
||||
let c = Channel { id: "UC1".into(), title: "First".into(), url: "u".into() };
|
||||
db.upsert_channels(&[c.clone()]).unwrap();
|
||||
db.upsert_channels(&[Channel { title: "Renamed".into(), ..c }]).unwrap();
|
||||
|
||||
let all = db.list_channels().unwrap();
|
||||
assert_eq!(all.len(), 1);
|
||||
assert_eq!(all[0].title, "Renamed");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn refresh_preserves_download_state() {
|
||||
let mut db = seeded();
|
||||
db.set_download_state("a", DownloadState::Done, None).unwrap();
|
||||
db.set_download_path("a", "/movies/a.mp4").unwrap();
|
||||
|
||||
// Simulate a later refresh returning updated metadata for the same video.
|
||||
let mut updated = vid("a", "UC1", 100, false);
|
||||
updated.title = "Retitled".into();
|
||||
updated.views = 999;
|
||||
db.upsert_videos(&[updated]).unwrap();
|
||||
|
||||
let feed = db.list_feed(&FeedFilter::default()).unwrap();
|
||||
let a = feed.iter().find(|f| f.id == "a").unwrap();
|
||||
assert_eq!(a.title, "Retitled", "metadata should update");
|
||||
assert_eq!(a.views, 999);
|
||||
assert_eq!(a.state, Some(DownloadState::Done), "download state must survive refresh");
|
||||
assert_eq!(a.path.as_deref(), Some("/movies/a.mp4"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn feed_is_sorted_newest_first() {
|
||||
let db = seeded();
|
||||
let feed = db.list_feed(&FeedFilter::default()).unwrap();
|
||||
let order: Vec<&str> = feed.iter().map(|f| f.id.as_str()).collect();
|
||||
assert_eq!(order, vec!["b", "c", "a"]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn feed_joins_channel_title() {
|
||||
let db = seeded();
|
||||
let feed = db.list_feed(&FeedFilter::default()).unwrap();
|
||||
let b = feed.iter().find(|f| f.id == "b").unwrap();
|
||||
assert_eq!(b.channel_title, "Beta");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn downloaded_only_filter_excludes_undownloaded() {
|
||||
let db = seeded();
|
||||
db.set_download_state("b", DownloadState::Done, None).unwrap();
|
||||
// A failed download must not count as available offline.
|
||||
db.set_download_state("a", DownloadState::Failed, Some("boom")).unwrap();
|
||||
|
||||
let feed = db
|
||||
.list_feed(&FeedFilter { downloaded_only: true, ..Default::default() })
|
||||
.unwrap();
|
||||
assert_eq!(feed.len(), 1);
|
||||
assert_eq!(feed[0].id, "b");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn hide_shorts_filter_excludes_shorts() {
|
||||
let db = seeded();
|
||||
let feed = db
|
||||
.list_feed(&FeedFilter { hide_shorts: true, ..Default::default() })
|
||||
.unwrap();
|
||||
assert_eq!(feed.len(), 2);
|
||||
assert!(feed.iter().all(|f| !f.is_short));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn channel_filter_scopes_to_one_channel() {
|
||||
let db = seeded();
|
||||
let feed = db
|
||||
.list_feed(&FeedFilter { channel_id: Some("UC1".into()), ..Default::default() })
|
||||
.unwrap();
|
||||
assert_eq!(feed.len(), 2);
|
||||
assert!(feed.iter().all(|f| f.channel_id == "UC1"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn search_matches_title_case_insensitively() {
|
||||
let db = seeded();
|
||||
let feed = db
|
||||
.list_feed(&FeedFilter { search: Some("title B".into()), ..Default::default() })
|
||||
.unwrap();
|
||||
assert_eq!(feed.len(), 1);
|
||||
assert_eq!(feed[0].id, "b");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn search_also_matches_channel_name() {
|
||||
let db = seeded();
|
||||
let feed = db
|
||||
.list_feed(&FeedFilter { search: Some("beta".into()), ..Default::default() })
|
||||
.unwrap();
|
||||
assert_eq!(feed.len(), 1);
|
||||
assert_eq!(feed[0].id, "b");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn channel_list_reports_video_and_download_counts() {
|
||||
let mut db = seeded();
|
||||
db.set_download_state("a", DownloadState::Done, None).unwrap();
|
||||
let chans = db.list_channels().unwrap();
|
||||
let alpha = chans.iter().find(|c| c.id == "UC1").unwrap();
|
||||
assert_eq!(alpha.video_count, 2);
|
||||
assert_eq!(alpha.downloaded_count, 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn progress_updates_are_recorded() {
|
||||
let db = seeded();
|
||||
db.set_download_state("a", DownloadState::Running, None).unwrap();
|
||||
db.set_download_progress("a", 50, Some(100), Some(50.0)).unwrap();
|
||||
let feed = db.list_feed(&FeedFilter::default()).unwrap();
|
||||
let a = feed.iter().find(|f| f.id == "a").unwrap();
|
||||
assert_eq!(a.pct, Some(50.0));
|
||||
assert_eq!(a.state, Some(DownloadState::Running));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn clearing_a_download_makes_it_undownloaded_again() {
|
||||
let db = seeded();
|
||||
db.set_download_state("a", DownloadState::Done, None).unwrap();
|
||||
db.clear_download("a").unwrap();
|
||||
let feed = db.list_feed(&FeedFilter::default()).unwrap();
|
||||
assert!(feed.iter().find(|f| f.id == "a").unwrap().state.is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn thumbs_needing_cache_are_reported_then_cleared() {
|
||||
let db = seeded();
|
||||
assert_eq!(db.videos_missing_thumbs(10).unwrap().len(), 3);
|
||||
db.set_thumb_path("a", "/cache/a.jpg").unwrap();
|
||||
assert_eq!(db.videos_missing_thumbs(10).unwrap().len(), 2);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user