From 8daa1b5c9dc1da4c180e58e19a421fdb7ed6bc0c Mon Sep 17 00:00:00 2001 From: Jeff Hughes Date: Thu, 30 Sep 2021 15:22:40 -0400 Subject: [PATCH] Add DB transactions when syncing --- src/db.rs | 91 +++++++++++++++++++++++++++++++++---------------------- 1 file changed, 54 insertions(+), 37 deletions(-) diff --git a/src/db.rs b/src/db.rs index d48f27c..3b87212 100644 --- a/src/db.rs +++ b/src/db.rs @@ -26,6 +26,7 @@ pub struct SyncResult { /// with this connection. #[derive(Debug)] pub struct Database { + path: PathBuf, conn: Option, } @@ -37,8 +38,9 @@ impl Database { std::fs::create_dir_all(&db_path) .with_context(|| "Unable to create subdirectory for database.")?; db_path.push("data.db"); - let conn = Connection::open(db_path)?; + let conn = Connection::open(&db_path)?; let db_conn = Database { + path: db_path, conn: Some(conn), }; db_conn.create()?; @@ -69,7 +71,7 @@ impl Database { // go here first, before we update the version // adding a column to capture episode guids - if db_version == Version::parse("1.2.1")? { + if db_version <= Version::parse("1.2.1")? { conn.execute("ALTER TABLE episodes ADD COLUMN guid TEXT;", params![]) .expect("Could not run database migrations."); } @@ -172,26 +174,33 @@ impl Database { /// Inserts a new podcast and list of podcast episodes into the /// database. pub fn insert_podcast(&self, podcast: PodcastNoId) -> Result { - let conn = self.conn.as_ref().expect("Error connecting to database."); - let mut stmt = conn.prepare_cached( - "INSERT INTO podcasts (title, url, description, author, + let mut conn = Connection::open(&self.path).expect("Error connecting to database."); + let tx = conn.transaction()?; + // let conn = self.conn.as_ref().expect("Error connecting to database."); + { + let mut stmt = tx.prepare_cached( + "INSERT INTO podcasts (title, url, description, author, explicit, last_checked) VALUES (?, ?, ?, ?, ?, ?);", - )?; - stmt.execute(params![ - podcast.title, - podcast.url, - podcast.description, - podcast.author, - podcast.explicit, - podcast.last_checked.timestamp() - ])?; + )?; + stmt.execute(params![ + podcast.title, + podcast.url, + podcast.description, + podcast.author, + podcast.explicit, + podcast.last_checked.timestamp() + ])?; + } - let mut stmt = conn.prepare_cached("SELECT id FROM podcasts WHERE url = ?")?; - let pod_id = stmt.query_row::(params![podcast.url], |row| row.get(0))?; + let pod_id; + { + let mut stmt = tx.prepare_cached("SELECT id FROM podcasts WHERE url = ?")?; + pod_id = stmt.query_row::(params![podcast.url], |row| row.get(0))?; + } let mut ep_ids = Vec::new(); for ep in podcast.episodes.iter().rev() { - let id = self.insert_episode(pod_id, ep)?; + let id = self.insert_episode(&tx, pod_id, ep)?; let new_ep = NewEpisode { id: id, pod_id: pod_id, @@ -201,6 +210,7 @@ impl Database { }; ep_ids.push(new_ep); } + tx.commit()?; return Ok(SyncResult { added: ep_ids, @@ -209,9 +219,12 @@ impl Database { } /// Inserts a podcast episode into the database. - pub fn insert_episode(&self, podcast_id: i64, episode: &EpisodeNoId) -> Result { - let conn = self.conn.as_ref().expect("Error connecting to database."); - + pub fn insert_episode( + &self, + conn: &Connection, + podcast_id: i64, + episode: &EpisodeNoId, + ) -> Result { let pubdate = episode.pubdate.map(|dt| dt.timestamp()); let mut stmt = conn.prepare_cached( @@ -283,21 +296,23 @@ impl Database { /// changed if necessary, and episodes are updated (modified episodes /// are updated, new episodes are inserted). pub fn update_podcast(&self, pod_id: i64, podcast: PodcastNoId) -> Result { - let conn = self.conn.as_ref().expect("Error connecting to database."); - let mut stmt = conn.prepare_cached( - "UPDATE podcasts SET title = ?, url = ?, description = ?, + { + let conn = self.conn.as_ref().expect("Error connecting to database."); + let mut stmt = conn.prepare_cached( + "UPDATE podcasts SET title = ?, url = ?, description = ?, author = ?, explicit = ?, last_checked = ? WHERE id = ?;", - )?; - stmt.execute(params![ - podcast.title, - podcast.url, - podcast.description, - podcast.author, - podcast.explicit, - podcast.last_checked.timestamp(), - pod_id, - ])?; + )?; + stmt.execute(params![ + podcast.title, + podcast.url, + podcast.description, + podcast.author, + podcast.explicit, + podcast.last_checked.timestamp(), + pod_id, + ])?; + } let result = self.update_episodes(pod_id, podcast.title, podcast.episodes)?; return Ok(result); @@ -317,8 +332,6 @@ impl Database { podcast_title: String, episodes: Vec, ) -> Result { - let conn = self.conn.as_ref().expect("Error connecting to database."); - let old_episodes = self.get_episodes(podcast_id, true)?; let mut old_ep_map = AHashMap::new(); for ep in old_episodes.iter() { @@ -327,6 +340,9 @@ impl Database { } } + let mut conn = Connection::open(&self.path).expect("Error connecting to database."); + let tx = conn.transaction()?; + let mut insert_ep = Vec::new(); let mut update_ep = Vec::new(); for new_ep in episodes.iter().rev() { @@ -371,7 +387,7 @@ impl Database { match existing_id { Some(id) => { if update { - let mut stmt = conn.prepare_cached( + let mut stmt = tx.prepare_cached( "UPDATE episodes SET title = ?, url = ?, guid = ?, description = ?, pubdate = ?, duration = ? WHERE id = ?;", @@ -389,7 +405,7 @@ impl Database { } } None => { - let id = self.insert_episode(podcast_id, new_ep)?; + let id = self.insert_episode(&tx, podcast_id, new_ep)?; let new_ep = NewEpisode { id: id, pod_id: podcast_id, @@ -401,6 +417,7 @@ impl Database { } } } + tx.commit()?; return Ok(SyncResult { added: insert_ep, updated: update_ep,