From bc53e0f730611da119df94799fe8a46af93048d1 Mon Sep 17 00:00:00 2001 From: rays Date: Fri, 18 Sep 2026 20:28:50 +0000 Subject: [PATCH] Postgres: pick the database by URL, copy-db, tests on both - IPX_DATABASE_URL (postgres://...) picks the database; unset, it is the SQLite file as before. Passwords are taken out of anything logged. - `ipx copy-db ` copies every table into the empty database the URL names, in one transaction, and moves the id counters past the copied ids. A copy of production went across in 14s with every count and column fingerprint identical. - With IPX_TEST_DATABASE_URL set, each test gets a Postgres schema of its own; all 79 pass on both databases. Fixtures write booleans as true/false. - Sorts say where an item with no value goes (NULLS FIRST going up, LAST going down): SQLite counts NULL as smallest, Postgres as largest, so "largest first" on Postgres led with every item that has no file. Tested on both. Co-Authored-By: Claude Opus 5 --- src/db.rs | 204 ++++++++++++++++++++++++++++++++++++++--------- src/main.rs | 21 ++++- src/retention.rs | 16 ++-- 3 files changed, 193 insertions(+), 48 deletions(-) diff --git a/src/db.rs b/src/db.rs index aa9f478..91fa768 100644 --- a/src/db.rs +++ b/src/db.rs @@ -67,14 +67,64 @@ async fn create_missing(orm: &sea_orm::DatabaseConnection) -> Result<()> { Ok(()) } +/// One table's rows, a page at a time in primary-key order, from one database to another. +async fn copy_table(from: &sea_orm::DatabaseConnection, to: &impl ConnectionTrait) -> Result +where + E: EntityTrait, + E::Model: sea_orm::IntoActiveModel + Send + Sync, + E::ActiveModel: ActiveModelTrait + Send, +{ + use sea_orm::{IntoActiveModel, Iterable, PrimaryKeyToColumn}; + let mut query = E::find(); + for key in E::PrimaryKey::iter() { + query = query.order_by_asc(key.into_column()); + } + let mut pages = query.paginate(from, 1000); + let mut n = 0; + while let Some(rows) = pages.fetch_and_next().await? { + n += rows.len() as u64; + // reset_all: every column written, the primary key included, not just the changed ones. + E::insert_many(rows.into_iter().map(|m| m.into_active_model().reset_all())) + .exec_without_returning(to) + .await?; + } + Ok(n) +} + +/// Where the database is: IPX_DATABASE_URL, a postgres:// URL, when it is set; otherwise the +/// SQLite file in the data directory, as it has always been. +pub fn location() -> String { + std::env::var("IPX_DATABASE_URL") + .ok() + .filter(|u| !u.trim().is_empty()) + .unwrap_or_else(|| crate::config::data_dir().join("state.db").display().to_string()) +} + +/// A URL fit for a log or an error: the password taken out. +fn redact(url: &str) -> String { + match (url.find("://"), url.rfind('@')) { + (Some(s), Some(at)) if at > s => match url[s + 3..at].find(':') { + Some(c) => format!("{}:***{}", &url[..s + 3 + c], &url[at..]), + None => url.to_owned(), + }, + _ => url.to_owned(), + } +} + +fn is_postgres(location: &str) -> bool { + location.starts_with("postgres://") || location.starts_with("postgresql://") +} + /// sqlx's defaults for SQLite are what ipx wants: foreign keys on, and a five-second wait for a /// lock, which is what rusqlite was set to. -async fn connect(path: &Path) -> Result { - let mut opts = sea_orm::ConnectOptions::new(format!("sqlite://{}?mode=rwc", path.display())); +async fn connect(location: &str) -> Result { + let url = + if is_postgres(location) { location.to_owned() } else { format!("sqlite://{location}?mode=rwc") }; + let mut opts = sea_orm::ConnectOptions::new(url); opts.sqlx_logging(false); sea_orm::Database::connect(opts) .await - .with_context(|| format!("opening {}", path.display())) + .with_context(|| format!("opening {}", redact(location))) } @@ -175,18 +225,23 @@ pub struct FeedSummary { } impl Db { - pub async fn open(path: &Path) -> Result { - if let Some(dir) = path.parent() { - std::fs::create_dir_all(dir) - .with_context(|| format!("creating {}", dir.display()))?; + /// Opens the database at `location` (see `location()`): a postgres:// URL, or a SQLite file. + pub async fn open(location: &str) -> Result { + if !is_postgres(location) + && let Some(dir) = Path::new(location).parent() + { + std::fs::create_dir_all(dir).with_context(|| format!("creating {}", dir.display()))?; } - let orm = connect(path).await?; - // WAL, so the healthcheck's `ipx status` reads while the daemon writes. It is a setting of - // the file, kept once made, and making it takes a lock that cannot wait out a busy - // daemon, so it is made only when the file is not already in WAL. - let mode = orm.query_one_raw(Statement::from_string(orm.get_database_backend(), "PRAGMA journal_mode")).await?; - if mode.and_then(|r| r.try_get_by_index::(0).ok()).as_deref() != Some("wal") { - orm.execute_unprepared("PRAGMA journal_mode = WAL").await.context("switching to WAL")?; + let orm = connect(location).await?; + if !is_postgres(location) { + // WAL, so the healthcheck's `ipx status` reads while the daemon writes. It is a + // setting of the file, kept once made, and making it takes a lock that cannot wait out + // a busy daemon, so it is made only when the file is not already in WAL. + let mode = + orm.query_one_raw(Statement::from_string(orm.get_database_backend(), "PRAGMA journal_mode")).await?; + if mode.and_then(|r| r.try_get_by_index::(0).ok()).as_deref() != Some("wal") { + orm.execute_unprepared("PRAGMA journal_mode = WAL").await.context("switching to WAL")?; + } } create_missing(&orm).await?; Ok(Self { @@ -196,17 +251,41 @@ impl Db { }) } - /// A fresh database for a test, in a file of its own: two connections to one ":memory:" - /// would be two databases. Its schema comes from the entities, as a real one's does. + /// A fresh, empty database for a test, its schema made from the entities as a real one's is. + /// A SQLite file of its own (two connections to one ":memory:" would be two databases); or, + /// with IPX_TEST_DATABASE_URL set, a Postgres schema of its own on that database, so the + /// same tests prove the SQL on both. #[cfg(test)] pub async fn memory() -> Result { static N: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0); - let path = std::env::temp_dir().join(format!( - "ipx-test-{}-{}.db", - std::process::id(), - N.fetch_add(1, std::sync::atomic::Ordering::Relaxed) - )); - let orm = connect(&path).await?; + let n = N.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + let pid = std::process::id(); + if let Some(url) = std::env::var("IPX_TEST_DATABASE_URL").ok().filter(|u| !u.is_empty()) { + let admin = connect(&url).await?; + // Schemas an earlier run left behind: anything of ours not made by this process. + let old = admin + .query_all_raw(Statement::from_string( + admin.get_database_backend(), + format!( + r"SELECT nspname FROM pg_namespace + WHERE nspname LIKE 'ipxt\_%' AND nspname NOT LIKE 'ipxt\_{pid}\_%'" + ), + )) + .await?; + for r in old { + let name: String = r.try_get_by_index(0)?; + admin.execute_unprepared(&format!("DROP SCHEMA IF EXISTS {name} CASCADE")).await?; + } + let schema = format!("ipxt_{pid}_{n}"); + admin.execute_unprepared(&format!("CREATE SCHEMA {schema}")).await?; + // Every connection in the pool starts in it, so each test sees only its own tables. + let sep = if url.contains('?') { '&' } else { '?' }; + let orm = connect(&format!("{url}{sep}options=-c%20search_path%3D{schema}")).await?; + create_missing(&orm).await?; + return Ok(Self { orm, tmp: None }); + } + let path = std::env::temp_dir().join(format!("ipx-test-{pid}-{n}.db")); + let orm = connect(&path.display().to_string()).await?; create_missing(&orm).await?; Ok(Self { orm, tmp: Some(path) }) } @@ -698,7 +777,10 @@ pub fn order_sql(col: &str, dir: &str, pinned_first: bool) -> String { "size" => "(SELECT max(x.length) FROM enclosures x WHERE x.feed_id = e.feed_id AND x.guid = e.guid)", _ => "coalesce(e.published, e.first_seen)", }; - let dir = if dir == "asc" { "ASC" } else { "DESC" }; + // Where an item with no value goes, said outright: SQLite counts a NULL as the smallest value + // and Postgres as the largest, so "largest first" on Postgres opened with every item that has + // no file. NULLS FIRST going up and LAST going down keeps what SQLite did. + let dir = if dir == "asc" { "ASC NULLS FIRST" } else { "DESC NULLS LAST" }; let pins = if pinned_first && col != "kept" { "coalesce(s.flagged, false) DESC, " } else { "" }; // The guid breaks what ties remain: SQLite's rowid did, and Postgres has none. format!("{pins}{expr} {dir}, coalesce(e.published, e.first_seen) DESC, e.guid DESC") @@ -882,6 +964,44 @@ impl Db { Ok(()) } + // ---- moving to another database ---- + + /// Copies every row of `from` into this database, which must be empty: the move from the + /// SQLite file to Postgres (issue #18). One transaction, so a copy that fails part-way leaves + /// nothing behind and can simply be run again. Returns each table's row count. + pub async fn copy_from(&self, from: &Db) -> Result> { + use crate::entity::*; + use sea_orm::TransactionTrait; + let held = users::Entity::find().count(&self.orm).await? + + feeds::Entity::find().count(&self.orm).await? + + entries::Entity::find().count(&self.orm).await?; + anyhow::ensure!(held == 0, "the database being copied into already holds rows; it has to be empty"); + let tx = self.orm.begin().await?; + // Users first: the tables that belong to a person refer to them. + let counts = vec![ + ("users", copy_table::(&from.orm, &tx).await?), + ("feeds", copy_table::(&from.orm, &tx).await?), + ("entries", copy_table::(&from.orm, &tx).await?), + ("enclosures", copy_table::(&from.orm, &tx).await?), + ("subscriptions", copy_table::(&from.orm, &tx).await?), + ("entry_state", copy_table::(&from.orm, &tx).await?), + ("sessions", copy_table::(&from.orm, &tx).await?), + ]; + if self.orm.get_database_backend() == sea_orm::DbBackend::Postgres { + // The copied ids came with the rows; the counters that hand out new ones start past + // them, or the next account or file would collide with one copied. + for table in ["users", "enclosures"] { + tx.execute_unprepared(&format!( + "SELECT setval(pg_get_serial_sequence('{table}', 'id'), \ + coalesce((SELECT max(id) FROM {table}), 0) + 1, false)" + )) + .await?; + } + } + tx.commit().await?; + Ok(counts) + } + // ---- hand-written SQL ---- // // For what reads better as SQL than as a query builder: joins, sums, upserts. Written to @@ -1598,7 +1718,7 @@ mod tests { async fn every_sort_column_runs_and_orders_both_ways() { let db = Db::memory().await.unwrap(); db.exec_for_test( - "INSERT INTO users (id, name, is_admin) VALUES (1,'ray',1); + "INSERT INTO users (id, name, is_admin) VALUES (1,'ray',true); INSERT INTO subscriptions (user_id, feed_id) VALUES (1,'f'),(1,'g'); INSERT INTO feeds (id, url, title) VALUES ('f','u','Zebra'),('g','v','Aardvark'); INSERT INTO entries (feed_id, guid, title, first_seen) VALUES @@ -1620,6 +1740,14 @@ mod tests { assert_eq!(order("feed", "asc").await, ["b", "c", "a"], "Aardvark, then Zebra's newest first"); assert_eq!(order("type", "asc").await, ["a", "b", "c"], "audio, image, video"); assert_eq!(order("size", "desc").await, ["c", "a", "b"]); + // An item with no file has no size or type: last going down, first going up, on both + // databases (they disagree about NULL unless told). + db.exec_for_test("INSERT INTO entries (feed_id, guid, title, first_seen) VALUES ('f','n','no file',50);") + .await + .unwrap(); + assert_eq!(order("size", "desc").await.last().map(String::as_str), Some("n")); + assert_eq!(order("type", "asc").await.first().map(String::as_str), Some("n")); + db.exec_for_test("DELETE FROM entries WHERE guid = 'n';").await.unwrap(); assert_eq!(order("published", "desc").await, ["c", "b", "a"]); db.set_entry_flag(1, "f", "a", EntryFlag::Flagged, true).await.unwrap(); assert_eq!(order("kept", "desc").await[0], "a"); @@ -1643,7 +1771,7 @@ mod tests { async fn deleting_a_shared_file_asks_about_everyone_else() { let db = Db::memory().await.unwrap(); db.exec_for_test( - "INSERT INTO users (id, name, is_admin) VALUES (1,'ray',1),(2,'sam',0),(3,'kit',0); + "INSERT INTO users (id, name, is_admin) VALUES (1,'ray',true),(2,'sam',false),(3,'kit',false); INSERT INTO subscriptions (user_id, feed_id) VALUES (1,'f'),(2,'f'),(3,'f'); INSERT INTO entries (feed_id, guid, first_seen) VALUES ('f','a',0); INSERT INTO enclosures (id, feed_id, guid, url, path, state) VALUES @@ -1669,7 +1797,7 @@ mod tests { async fn read_state_belongs_to_one_person() { let db = Db::memory().await.unwrap(); db.exec_for_test( - "INSERT INTO users (id, name, is_admin) VALUES (1,'ray',1),(2,'sam',0); + "INSERT INTO users (id, name, is_admin) VALUES (1,'ray',true),(2,'sam',false); INSERT INTO entries (feed_id, guid, title, first_seen) VALUES ('f','a','One',100),('f','b','Two',200);", ).await @@ -1751,7 +1879,7 @@ mod tests { // Cutting this along with the dead read columns left the browser suite's admin with an // empty sidebar: it is how a fresh install's first account gets config.toml's feeds. let db = Db::memory().await.unwrap(); - db.exec_for_test("INSERT INTO users (id, name, is_admin) VALUES (1,'admin',1);").await + db.exec_for_test("INSERT INTO users (id, name, is_admin) VALUES (1,'admin',true);").await .unwrap(); let subs = async || db.i64s_for_test("SELECT count(*) FROM subscriptions").await[0]; let catalogue = ["a".to_string(), "b".to_string()]; @@ -1781,19 +1909,19 @@ mod tests { INSERT INTO enclosures (id, feed_id, guid, url, path, state) VALUES (1,'f','b','u1','/tmp/b','done'); -- Read and starred belong to a person now, so say which one. - INSERT INTO users (id, name, is_admin) VALUES (7,'reader',1); + INSERT INTO users (id, name, is_admin) VALUES (7,'reader',true); INSERT INTO entry_state (user_id, feed_id, guid, read, flagged, position) VALUES - (7,'f','b',1,0,0), - (7,'f','c',1,1,0), + (7,'f','b',true,false,0), + (7,'f','c',true,true,0), -- Started, length unknown: this is Currently Listening. - (7,'f','d',0,0,42), + (7,'f','d',false,false,42), -- 42 of 45 seconds is past the 90% the player calls finished. - (7,'f','e',1,0,42), + (7,'f','e',true,false,42), -- Barely touched (opened, closed within seconds): not Currently Listening. - (7,'f','g',0,0,3), + (7,'f','g',false,false,3), -- Opened, and so read, but 42 of 900 seconds in: still Currently Listening. -- Filtering on read hid exactly these (issue #14). - (7,'f','h',1,0,42);", + (7,'f','h',true,false,42);", ).await .unwrap(); @@ -1877,13 +2005,13 @@ mod tests { async fn a_show_takes_over_what_its_creator_held() { let db = Db::memory().await.unwrap(); db.exec_for_test( - "INSERT INTO users (id, name, is_admin) VALUES (1,'ray',1); + "INSERT INTO users (id, name, is_admin) VALUES (1,'ray',true); INSERT INTO enclosures (id, feed_id, guid, url, state, path, last_error) VALUES (1,'creator','a','u1','done','/x/a.mp3',NULL), (2,'creator','b','u2','skipped',NULL,'explicit'), (3,'creator','c','u3','skipped',NULL,'explicit'), (4,'other','d','u4','skipped',NULL,'torrents disabled'); - INSERT INTO entry_state (user_id, feed_id, guid, read) VALUES (1,'creator','a',1);", + INSERT INTO entry_state (user_id, feed_id, guid, read) VALUES (1,'creator','a',true);", ).await .unwrap(); db.adopt("creator", "show", &[("a", "u1"), ("b", "u2"), ("d", "u4")]).await.unwrap(); @@ -1909,9 +2037,9 @@ mod tests { async fn a_feed_in_a_group_follows_your_settings_on_the_group() { let db = Db::memory().await.unwrap(); db.exec_for_test( - "INSERT INTO users (id, name, is_admin) VALUES (1,'ray',1),(2,'sam',0); + "INSERT INTO users (id, name, is_admin) VALUES (1,'ray',true),(2,'sam',false); INSERT INTO subscriptions (user_id, feed_id, allow_explicit) VALUES - (1,'group',1),(1,'show',NULL),(2,'group',1),(2,'show',0);", + (1,'group',true),(1,'show',NULL),(2,'group',true),(2,'show',false);", ).await .unwrap(); let explicit = async |group| -> Vec> { diff --git a/src/main.rs b/src/main.rs index 595c3a4..9bcbd12 100644 --- a/src/main.rs +++ b/src/main.rs @@ -37,6 +37,12 @@ struct Cli { enum Command { /// Show configured feeds and their state List, + /// Copy everything from a SQLite state.db into the database IPX_DATABASE_URL names, which + /// must be empty: the one-off move to Postgres + CopyDb { + /// The SQLite file to copy from + from: PathBuf, + }, /// Scan feeds for new entries Fetch { /// Only this feed id @@ -180,7 +186,7 @@ async fn main() -> Result<()> { let config_path = cli.config.clone().unwrap_or_else(config::config_path); let cfg = config::Config::load(&config_path)?; - let db = db::Db::open(&config::data_dir().join("state.db")).await?; + let db = db::Db::open(&db::location()).await?; // A daemon owns the state; don't have two processes downloading the same thing. let wire_cmd = match &cli.command { @@ -195,7 +201,8 @@ async fn main() -> Result<()> { | Command::Add { .. } | Command::Rm { .. } | Command::Import { .. } - | Command::Export { .. } => None, + | Command::Export { .. } + | Command::CopyDb { .. } => None, }; if let Some(cmd) = &wire_cmd && !cli.local @@ -229,6 +236,7 @@ async fn main() -> Result<()> { Command::User { cmd } => user_cmd(&ctx, cmd).await, Command::Import { file } => import(&ctx, &config_path, &file).await, Command::Export { file } => export(&ctx, &file).await, + Command::CopyDb { from } => copy_db(&ctx, &from).await, _ => run(&ctx, wire_cmd.expect("only List and Daemon have no wire form")).await, } } @@ -770,6 +778,15 @@ pub fn collect_outlines(outlines: &[opml::Outline], out: &mut Vec<(String, Strin } } +async fn copy_db(ctx: &Ctx, from: &std::path::Path) -> Result<()> { + anyhow::ensure!(from.exists(), "{} does not exist", from.display()); + let source = db::Db::open(&from.display().to_string()).await?; + for (table, n) in ctx.db.copy_from(&source).await? { + println!("{table:14} {n}"); + } + Ok(()) +} + async fn export(ctx: &Ctx, file: &std::path::Path) -> Result<()> { let mut doc = opml::OPML::default(); doc.head = Some(opml::Head { diff --git a/src/retention.rs b/src/retention.rs index e58f2eb..009f7cd 100644 --- a/src/retention.rs +++ b/src/retention.rs @@ -154,7 +154,7 @@ mod tests { // One file serves both subscribers, so it takes both of them to release it. let db = Db::memory().await.unwrap(); db.exec_for_test( - "INSERT INTO users (id, name, is_admin) VALUES (1,'ray',1),(2,'sam',0); + "INSERT INTO users (id, name, is_admin) VALUES (1,'ray',true),(2,'sam',false); INSERT INTO subscriptions (user_id, feed_id) VALUES (1,'f'),(2,'f'); INSERT INTO entries (feed_id, guid, first_seen) VALUES ('f', 'keep', 0), @@ -163,11 +163,11 @@ mod tests { ('f', 'read', 0); -- Starred by one of the two, so it stays whatever the other thinks. INSERT INTO entry_state (user_id, feed_id, guid, read, flagged) VALUES - (1, 'f', 'keep', 1, 1), - (2, 'f', 'keep', 1, 0), - (1, 'f', 'half', 1, 0), - (1, 'f', 'read', 1, 0), - (2, 'f', 'read', 1, 0); + (1, 'f', 'keep', true, true), + (2, 'f', 'keep', true, false), + (1, 'f', 'half', true, false), + (1, 'f', 'read', true, false), + (2, 'f', 'read', true, false); INSERT INTO enclosures (id, feed_id, guid, url, path, bytes_done, state, downloaded_at) VALUES (1, 'f', 'keep', 'u1', '/tmp/keep', 10, 'done', 10), (2, 'f', 'half', 'u2', '/tmp/half', 10, 'done', 20), @@ -189,8 +189,8 @@ mod tests { async fn prune_keeps_entries_that_still_have_a_file() { let db = Db::memory().await.unwrap(); db.exec_for_test( - "INSERT INTO users (id, name, is_admin) VALUES (1,'ray',1); - INSERT INTO entry_state (user_id, feed_id, guid, flagged) VALUES (1,'f','flagged',1); + "INSERT INTO users (id, name, is_admin) VALUES (1,'ray',true); + INSERT INTO entry_state (user_id, feed_id, guid, flagged) VALUES (1,'f','flagged',true); INSERT INTO entries (feed_id, guid, first_seen) VALUES ('f', 'has-file', 100), ('f', 'no-file', 100),