move db upgrade logic into db crate

This allows shrinking db's API surface.

db/dir.rs

@@ -226,8 +226,7 @@ impl SampleFileDir {
         Ok(meta)
     }
 
-    // TODO: this should be exposed only to the db layer.
+    pub(crate) fn write_meta(&self, meta: &schema::DirMeta) -> Result<(), Error> {
-    pub fn write_meta(&self, meta: &schema::DirMeta) -> Result<(), Error> {
         let (tmp_path, final_path) = unsafe {
             (ffi::CStr::from_ptr("meta.tmp\0".as_ptr() as *const c_char),
              ffi::CStr::from_ptr("meta\0".as_ptr() as *const c_char))

db/lib.rs

@@ -50,7 +50,8 @@ pub mod db;
 pub mod dir;
 mod raw;
 pub mod recording;
-pub mod schema;
+mod schema;
+pub mod upgrade;
 
 // This is only for #[cfg(test)], but it's also used by the dependent crate, and it appears that
 // #[cfg(test)] is not passed on to dependencies.

db/upgrade/mod.rs

@@ -0,0 +1,111 @@
+// This file is part of Moonfire NVR, a security camera digital video recorder.
+// Copyright (C) 2016 Scott Lamb <slamb@slamb.org>
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// In addition, as a special exception, the copyright holders give
+// permission to link the code of portions of this program with the
+// OpenSSL library under certain conditions as described in each
+// individual source file, and distribute linked combinations including
+// the two.
+//
+// You must obey the GNU General Public License in all respects for all
+// of the code used other than OpenSSL. If you modify file(s) with this
+// exception, you may extend this exception to your version of the
+// file(s), but you are not obligated to do so. If you do not wish to do
+// so, delete this exception statement from your version. If you delete
+// this exception statement from all source files in the program, then
+// also delete it here.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+/// Upgrades the database schema.
+///
+/// See `guide/schema.md` for more information.
+
+use db;
+use failure::Error;
+use rusqlite;
+
+mod v0_to_v1;
+mod v1_to_v2;
+mod v2_to_v3;
+
+const UPGRADE_NOTES: &'static str =
+    concat!("upgraded using moonfire-db ", env!("CARGO_PKG_VERSION"));
+
+pub trait Upgrader {
+    fn in_tx(&mut self, &rusqlite::Transaction) -> Result<(), Error> { Ok(()) }
+    fn post_tx(&mut self) -> Result<(), Error> { Ok(()) }
+}
+
+#[derive(Debug)]
+pub struct Args<'a> {
+    pub flag_sample_file_dir: Option<&'a str>,
+    pub flag_preset_journal: &'a str,
+    pub flag_no_vacuum: bool,
+}
+
+fn set_journal_mode(conn: &rusqlite::Connection, requested: &str) -> Result<(), Error> {
+    assert!(!requested.contains(';'));  // quick check for accidental sql injection.
+    let actual = conn.query_row(&format!("pragma journal_mode = {}", requested), &[],
+                                |row| row.get_checked::<_, String>(0))??;
+    info!("...database now in journal_mode {} (requested {}).", actual, requested);
+    Ok(())
+}
+
+pub fn run(args: &Args, conn: &mut rusqlite::Connection) -> Result<(), Error> {
+    let upgraders = [
+        v0_to_v1::new,
+        v1_to_v2::new,
+        v2_to_v3::new,
+    ];
+
+    {
+        assert_eq!(upgraders.len(), db::EXPECTED_VERSION as usize);
+        let old_ver =
+            conn.query_row("select max(id) from version", &[], |row| row.get_checked(0))??;
+        if old_ver > db::EXPECTED_VERSION {
+            bail!("Database is at version {}, later than expected {}",
+                  old_ver, db::EXPECTED_VERSION);
+        } else if old_ver < 0 {
+            bail!("Database is at negative version {}!", old_ver);
+        }
+        info!("Upgrading database from version {} to version {}...", old_ver, db::EXPECTED_VERSION);
+        set_journal_mode(&conn, args.flag_preset_journal).unwrap();
+        for ver in old_ver .. db::EXPECTED_VERSION {
+            info!("...from version {} to version {}", ver, ver + 1);
+            let mut u = upgraders[ver as usize](&args)?;
+            let tx = conn.transaction()?;
+            u.in_tx(&tx)?;
+            tx.execute(r#"
+                insert into version (id, unix_time, notes)
+                             values (?, cast(strftime('%s', 'now') as int32), ?)
+            "#, &[&(ver + 1), &UPGRADE_NOTES])?;
+            tx.commit()?;
+            u.post_tx()?;
+        }
+    }
+
+    // WAL is the preferred journal mode for normal operation; it reduces the number of syncs
+    // without compromising safety.
+    set_journal_mode(&conn, "wal").unwrap();
+    if !args.flag_no_vacuum {
+        info!("...vacuuming database after upgrade.");
+        conn.execute_batch(r#"
+            pragma page_size = 16384;
+            vacuum;
+        "#).unwrap();
+    }
+    info!("...done.");
+    Ok(())
+}

db/upgrade/v0_to_v1.rs

@@ -30,11 +30,11 @@
 
 /// Upgrades a version 0 schema to a version 1 schema.
 
-use db::{self, recording};
+use db;
 use failure::Error;
+use recording;
 use rusqlite;
 use std::collections::HashMap;
-use strutil;
 
 pub struct U;
 
@@ -169,14 +169,14 @@ fn fill_recording(tx: &rusqlite::Transaction) -> Result<HashMap<i32, CameraState
         let video_samples: i32 = row.get_checked(5)?;
         let video_sync_samples: i32 = row.get_checked(6)?;
         let video_sample_entry_id: i32 = row.get_checked(7)?;
-        let sample_file_uuid: Vec<u8> = row.get_checked(8)?;
+        let sample_file_uuid: db::FromSqlUuid = row.get_checked(8)?;
         let sample_file_sha1: Vec<u8> = row.get_checked(9)?;
         let video_index: Vec<u8> = row.get_checked(10)?;
         let old_id: i32 = row.get_checked(11)?;
         let trailing_zero = has_trailing_zero(&video_index).unwrap_or_else(|e| {
             warn!("recording {}/{} (sample file {}, formerly recording {}) has corrupt \
                   video_index: {}",
-                  camera_id, composite_id & 0xFFFF, strutil::hex(&sample_file_uuid), old_id, e);
+                  camera_id, composite_id & 0xFFFF, sample_file_uuid.0, old_id, e);
             false
         });
         let run_id = match camera_state.current_run {
@@ -198,7 +198,7 @@ fn fill_recording(tx: &rusqlite::Transaction) -> Result<HashMap<i32, CameraState
         ])?;
         insert2.execute_named(&[
             (":composite_id", &composite_id),
-            (":sample_file_uuid", &sample_file_uuid),
+            (":sample_file_uuid", &&sample_file_uuid.0.as_bytes()[..]),
             (":sample_file_sha1", &sample_file_sha1),
             (":video_index", &video_index),
         ])?;

db/upgrade/v1_to_v2.rs

@@ -30,11 +30,11 @@
 
 /// Upgrades a version 1 schema to a version 2 schema.
 
-use db::schema::DirMeta;
+use dir;
-use db::dir;
 use failure::Error;
-use std::fs;
 use rusqlite;
+use schema::DirMeta;
+use std::fs;
 use uuid::Uuid;
 
 pub struct U<'a> {
@@ -45,7 +45,6 @@ pub struct U<'a> {
 pub fn new<'a>(args: &'a super::Args) -> Result<Box<super::Upgrader + 'a>, Error> {
     let sample_file_path =
         args.flag_sample_file_dir
-            .as_ref()
             .ok_or_else(|| format_err!("--sample-file-dir required when upgrading from \
                                         schema version 1 to 2."))?;
     Ok(Box::new(U { sample_file_path, dir_meta: None }))

db/upgrade/v2_to_v3.rs

@@ -30,12 +30,12 @@
 
 /// Upgrades a version 2 schema to a version 3 schema.
 
-use db::{self, dir, FromSqlUuid};
+use db::{self, FromSqlUuid};
+use dir;
 use failure::Error;
 use libc;
 use std::io::{self, Write};
 use std::mem;
-//use std::rc::Rc;
 use rusqlite;
 use uuid::Uuid;
 
@@ -45,31 +45,6 @@ pub fn new<'a>(_args: &'a super::Args) -> Result<Box<super::Upgrader + 'a>, Erro
     Ok(Box::new(U))
 }
 
-/*fn build_stream_to_dir(dir: &dir::Fd, tx: &rusqlite::Transaction)
-                           -> Result<Vec<Option<Rc<dir::Fd>>>, Error> {
-    let mut v = Vec::new();
-    let max_id: u32 = tx.query_row("select max(id) from stream", &[], |r| r.get_checked(0))??;
-    v.resize((max_id + 1) as usize, None);
-    let mut stmt = tx.prepare(r#"
-        select
-          stream.id,
-          camera.uuid,
-          stream.type
-        from
-          camera join stream on (camera.id = stream.camera_id)
-    "#)?;
-    let mut rows = stmt.query(&[])?;
-    while let Some(row) = rows.next() {
-        let row = row?;
-        let id: i32 = row.get_checked(0)?;
-        let uuid: FromSqlUuid = row.get_checked(1)?;
-        let type_: String = row.get_checked(2)?;
-        v[id as usize] =
-            Some(Rc::new(dir::Fd::open(Some(dir), &format!("{}-{}", uuid.0, type_), true)?));
-    }
-    Ok(v)
-}*/
-
 /// Gets a pathname for a sample file suitable for passing to open or unlink.
 fn get_uuid_pathname(uuid: Uuid) -> [libc::c_char; 37] {
     let mut buf = [0u8; 37];
@@ -87,30 +62,6 @@ fn get_id_pathname(id: db::CompositeId) -> [libc::c_char; 17] {
 
 impl super::Upgrader for U {
     fn in_tx(&mut self, tx: &rusqlite::Transaction) -> Result<(), Error> {
-        /*let (meta, path) = tx.query_row(r#"
-            select
-              meta.uuid,
-              dir.path,
-              dir.uuid,
-              dir.last_complete_open_id,
-              open.uuid
-            from
-              meta cross join sample_file_dir dir
-              join open on (dir.last_complete_open_id = open.id)
-        "#, |row| -> Result<_, Error> {
-            let mut meta = DirMeta::new();
-            let db_uuid: FromSqlUuid = row.get_checked(0)?;
-            let path: String = row.get_checked(1)?;
-            let dir_uuid: FromSqlUuid = row.get_checked(2)?;
-            let open_uuid: FromSqlUuid = row.get_checked(4)?;
-            meta.db_uuid.extend_from_slice(&db_uuid.0.as_bytes()[..]);
-            meta.dir_uuid.extend_from_slice(&dir_uuid.0.as_bytes()[..]);
-            let open = meta.mut_last_complete_open();
-            open.id = row.get_checked(3)?;
-            open.uuid.extend_from_slice(&open_uuid.0.as_bytes()[..]);
-            Ok((meta, path))
-        })??;*/
-
         let path: String = tx.query_row(r#"
             select path from sample_file_dir
         "#, &[], |row| { row.get_checked(0) })??;

src/cmds/upgrade/mod.rs

@@ -34,11 +34,6 @@
 
 use db;
 use failure::Error;
-use rusqlite;
-
-mod v0_to_v1;
-mod v1_to_v2;
-mod v2_to_v3;
 
 const USAGE: &'static str = r#"
 Upgrade to the latest database schema.
@@ -62,14 +57,6 @@ Options:
     --no-vacuum            Skips the normal post-upgrade vacuum operation.
 "#;
 
-const UPGRADE_NOTES: &'static str =
-    concat!("upgraded using moonfire-nvr ", env!("CARGO_PKG_VERSION"));
-
-pub trait Upgrader {
-    fn in_tx(&mut self, &rusqlite::Transaction) -> Result<(), Error> { Ok(()) }
-    fn post_tx(&mut self) -> Result<(), Error> { Ok(()) }
-}
-
 #[derive(Debug, Deserialize)]
 pub struct Args {
     flag_db_dir: String,
@@ -78,60 +65,13 @@ pub struct Args {
     flag_no_vacuum: bool,
 }
 
-fn set_journal_mode(conn: &rusqlite::Connection, requested: &str) -> Result<(), Error> {
-    assert!(!requested.contains(';'));  // quick check for accidental sql injection.
-    let actual = conn.query_row(&format!("pragma journal_mode = {}", requested), &[],
-                                |row| row.get_checked::<_, String>(0))??;
-    info!("...database now in journal_mode {} (requested {}).", actual, requested);
-    Ok(())
-}
-
 pub fn run() -> Result<(), Error> {
     let args: Args = super::parse_args(USAGE)?;
     let (_db_dir, mut conn) = super::open_conn(&args.flag_db_dir, super::OpenMode::ReadWrite)?;
 
-    let upgraders = [
+    db::upgrade::run(&db::upgrade::Args {
-        v0_to_v1::new,
+        flag_sample_file_dir: args.flag_sample_file_dir.as_ref().map(|s| s.as_str()),
-        v1_to_v2::new,
+        flag_preset_journal: &args.flag_preset_journal,
-        v2_to_v3::new,
+        flag_no_vacuum: args.flag_no_vacuum,
-    ];
+    }, &mut conn)
-
-    {
-        assert_eq!(upgraders.len(), db::EXPECTED_VERSION as usize);
-        let old_ver =
-            conn.query_row("select max(id) from version", &[], |row| row.get_checked(0))??;
-        if old_ver > db::EXPECTED_VERSION {
-            bail!("Database is at version {}, later than expected {}",
-                  old_ver, db::EXPECTED_VERSION);
-        } else if old_ver < 0 {
-            bail!("Database is at negative version {}!", old_ver);
-        }
-        info!("Upgrading database from version {} to version {}...", old_ver, db::EXPECTED_VERSION);
-        set_journal_mode(&conn, &args.flag_preset_journal).unwrap();
-        for ver in old_ver .. db::EXPECTED_VERSION {
-            info!("...from version {} to version {}", ver, ver + 1);
-            let mut u = upgraders[ver as usize](&args)?;
-            let tx = conn.transaction()?;
-            u.in_tx(&tx)?;
-            tx.execute(r#"
-                insert into version (id, unix_time, notes)
-                             values (?, cast(strftime('%s', 'now') as int32), ?)
-            "#, &[&(ver + 1), &UPGRADE_NOTES])?;
-            tx.commit()?;
-            u.post_tx()?;
-        }
-    }
-
-    // WAL is the preferred journal mode for normal operation; it reduces the number of syncs
-    // without compromising safety.
-    set_journal_mode(&conn, "wal").unwrap();
-    if !args.flag_no_vacuum {
-        info!("...vacuuming database after upgrade.");
-        conn.execute_batch(r#"
-            pragma page_size = 16384;
-            vacuum;
-        "#).unwrap();
-    }
-    info!("...done.");
-    Ok(())
 }