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 <state.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 <noreply@anthropic.com>

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<E>(from: &sea_orm::DatabaseConnection, to: &impl ConnectionTrait) -> Result<u64>
+where
+    E: EntityTrait,
+    E::Model: sea_orm::IntoActiveModel<E::ActiveModel> + Send + Sync,
+    E::ActiveModel: ActiveModelTrait<Entity = E> + 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<sea_orm::DatabaseConnection> {
+async fn connect(location: &str) -> Result<sea_orm::DatabaseConnection> {
-    let mut opts = sea_orm::ConnectOptions::new(format!("sqlite://{}?mode=rwc", path.display()));
+    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<Self> {
+    /// Opens the database at `location` (see `location()`): a postgres:// URL, or a SQLite file.
-        if let Some(dir) = path.parent() {
+    pub async fn open(location: &str) -> Result<Self> {
-            std::fs::create_dir_all(dir)
+        if !is_postgres(location)
-                .with_context(|| format!("creating {}", dir.display()))?;
+            && let Some(dir) = Path::new(location).parent()
+        {
+            std::fs::create_dir_all(dir).with_context(|| format!("creating {}", dir.display()))?;
         }
-        let orm = connect(path).await?;
+        let orm = connect(location).await?;
-        // WAL, so the healthcheck's `ipx status` reads while the daemon writes. It is a setting of
+        if !is_postgres(location) {
-        // the file, kept once made, and making it takes a lock that cannot wait out a busy
+            // WAL, so the healthcheck's `ipx status` reads while the daemon writes. It is a
-        // daemon, so it is made only when the file is not already in WAL.
+            // setting of the file, kept once made, and making it takes a lock that cannot wait out
-        let mode = orm.query_one_raw(Statement::from_string(orm.get_database_backend(), "PRAGMA journal_mode")).await?;
+            // a busy daemon, so it is made only when the file is not already in WAL.
-        if mode.and_then(|r| r.try_get_by_index::<String>(0).ok()).as_deref() != Some("wal") {
+            let mode =
-            orm.execute_unprepared("PRAGMA journal_mode = WAL").await.context("switching to WAL")?;
+                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::<String>(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:"
+    /// A fresh, empty database for a test, its schema made from the entities as a real one's is.
-    /// would be two databases. Its schema comes from the entities, as a real one's does.
+    /// 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<Self> {
         static N: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0);
-        let path = std::env::temp_dir().join(format!(
+        let n = N.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
-            "ipx-test-{}-{}.db",
+        let pid = std::process::id();
-            std::process::id(),
+        if let Some(url) = std::env::var("IPX_TEST_DATABASE_URL").ok().filter(|u| !u.is_empty()) {
-            N.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
+            let admin = connect(&url).await?;
-        ));
+            // Schemas an earlier run left behind: anything of ours not made by this process.
-        let orm = connect(&path).await?;
+            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<Vec<(&'static str, u64)>> {
+        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::<users::Entity>(&from.orm, &tx).await?),
+            ("feeds", copy_table::<feeds::Entity>(&from.orm, &tx).await?),
+            ("entries", copy_table::<entries::Entity>(&from.orm, &tx).await?),
+            ("enclosures", copy_table::<enclosures::Entity>(&from.orm, &tx).await?),
+            ("subscriptions", copy_table::<subscriptions::Entity>(&from.orm, &tx).await?),
+            ("entry_state", copy_table::<entry_state::Entity>(&from.orm, &tx).await?),
+            ("sessions", copy_table::<sessions::Entity>(&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','b',true,false,0),
-                (7,'f','c',1,1,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<Option<bool>> {

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 {

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),
+                (1, 'f', 'keep',   true, true),
-                (2, 'f', 'keep',   1, 0),
+                (2, 'f', 'keep',   true, false),
-                (1, 'f', 'half',   1, 0),
+                (1, 'f', 'half',   true, false),
-                (1, 'f', 'read',   1, 0),
+                (1, 'f', 'read',   true, false),
-                (2, 'f', 'read',   1, 0);
+                (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 users (id, name, is_admin) VALUES (1,'ray',true);
-             INSERT INTO entry_state (user_id, feed_id, guid, flagged) VALUES (1,'f','flagged',1);
+             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),