pito_sync/
store.rs

1use std::fs::{self, OpenOptions};
2use std::os::unix::fs::{OpenOptionsExt, PermissionsExt};
3use std::path::{Path, PathBuf};
4use std::time::Duration;
5
6use rusqlite::{Connection, OpenFlags, TransactionBehavior};
7
8pub const SYNC_VERSION: i64 = 1;
9
10pub(crate) const STRIDE: i64 = 1000;
11
12pub const MAX_VERSION: i64 = (i32::MAX as i64 - (STRIDE - 1)) / STRIDE;
13
14const _: () = assert!(SYNC_VERSION > 0 && SYNC_VERSION < STRIDE);
15
16const SYNC: &str = "CREATE TABLE sync (
17  source TEXT PRIMARY KEY,
18  started_at INTEGER,
19  finished_at INTEGER,
20  ok INTEGER,
21  error TEXT,
22  cursor TEXT,
23  rows INTEGER,
24  retry_after INTEGER
25)";
26
27#[derive(Debug, Clone, PartialEq, Eq)]
28pub struct Spec {
29    pub path: PathBuf,
30    pub version: i64,
31    pub schema: &'static [&'static str],
32}
33
34impl Spec {
35    pub fn new(path: impl Into<PathBuf>, version: i64, schema: &'static [&'static str]) -> Self {
36        Spec {
37            path: path.into(),
38            version,
39            schema,
40        }
41    }
42
43    pub fn user_version(&self) -> i64 {
44        self.version
45            .saturating_mul(STRIDE)
46            .saturating_add(SYNC_VERSION)
47    }
48
49    fn stamp(&self) -> Result<i64, String> {
50        if (0..=MAX_VERSION).contains(&self.version) {
51            Ok(self.user_version())
52        } else {
53            Err(format!(
54                "{}: version {} is outside 0..={MAX_VERSION}",
55                self.path.display(),
56                self.version
57            ))
58        }
59    }
60}
61
62fn private(path: &Path, mode: u32) -> Result<(), String> {
63    let meta = fs::metadata(path).map_err(|e| format!("cannot read {}: {e}", path.display()))?;
64    if meta.permissions().mode() & 0o777 != mode {
65        fs::set_permissions(path, fs::Permissions::from_mode(mode))
66            .map_err(|e| format!("cannot protect {}: {e}", path.display()))?;
67    }
68    Ok(())
69}
70
71fn pragmas(conn: &Connection) -> rusqlite::Result<()> {
72    conn.busy_timeout(Duration::from_millis(5000))?;
73    conn.pragma_update(None, "journal_mode", "WAL")?;
74    conn.pragma_update(None, "synchronous", "NORMAL")?;
75    conn.pragma_update(None, "foreign_keys", "ON")
76}
77
78pub fn version(conn: &Connection) -> rusqlite::Result<i64> {
79    conn.query_row("PRAGMA user_version", [], |row| row.get(0))
80}
81
82fn remove(path: &Path) -> Result<(), String> {
83    for suffix in ["", "-wal", "-shm"] {
84        let mut name = path.as_os_str().to_owned();
85        name.push(suffix);
86        let file = PathBuf::from(name);
87        match fs::remove_file(&file) {
88            Ok(()) => {}
89            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
90            Err(e) => return Err(format!("cannot remove {}: {e}", file.display())),
91        }
92    }
93    Ok(())
94}
95
96fn build(conn: &mut Connection, schema: &[&str], stamp: i64) -> rusqlite::Result<()> {
97    let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
98    tx.execute_batch(SYNC)?;
99    for statement in schema {
100        tx.execute_batch(statement)?;
101    }
102    tx.pragma_update(None, "user_version", stamp)?;
103    tx.commit()
104}
105
106fn connect(path: &Path) -> Result<Connection, String> {
107    let shown = path.display();
108    OpenOptions::new()
109        .create(true)
110        .append(true)
111        .mode(0o600)
112        .open(path)
113        .map_err(|e| format!("cannot create {shown}: {e}"))?;
114    private(path, 0o600)?;
115    let conn = Connection::open(path).map_err(|e| format!("cannot open {shown}: {e}"))?;
116    pragmas(&conn).map_err(|e| format!("cannot prepare {shown}: {e}"))?;
117    Ok(conn)
118}
119
120pub fn open(spec: &Spec) -> Result<Connection, String> {
121    let stamp = spec.stamp()?;
122    let path = spec.path.as_path();
123    let shown = path.display();
124    if let Some(dir) = path.parent().filter(|dir| !dir.as_os_str().is_empty()) {
125        fs::create_dir_all(dir).map_err(|e| format!("cannot write {}: {e}", dir.display()))?;
126        private(dir, 0o700)?;
127    }
128    let found = connect(path).and_then(|conn| {
129        version(&conn)
130            .map(|found| (conn, found))
131            .map_err(|e| format!("cannot read {shown}: {e}"))
132    });
133    match found {
134        Ok((conn, found)) if found == stamp => return Ok(conn),
135        other => drop(other),
136    }
137    remove(path)?;
138    let mut conn = connect(path)?;
139    build(&mut conn, spec.schema, stamp).map_err(|e| format!("cannot build {shown}: {e}"))?;
140    Ok(conn)
141}
142
143pub fn reader(spec: &Spec) -> Result<Option<Connection>, String> {
144    let stamp = spec.stamp()?;
145    let path = spec.path.as_path();
146    if !path.exists() {
147        return Ok(None);
148    }
149    let shown = path.display();
150    let conn = Connection::open_with_flags(
151        path,
152        OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX,
153    )
154    .map_err(|e| format!("cannot open {shown}: {e}"))?;
155    conn.busy_timeout(Duration::from_millis(5000))
156        .map_err(|e| format!("cannot read {shown}: {e}"))?;
157    match version(&conn) {
158        Ok(found) if found == stamp => Ok(Some(conn)),
159        Ok(_) => Ok(None),
160        Err(e) => Err(format!("cannot read {shown}: {e}")),
161    }
162}
163
164pub fn memory(spec: &Spec) -> Result<Connection, String> {
165    let stamp = spec.stamp()?;
166    let mut conn = Connection::open_in_memory().map_err(|e| e.to_string())?;
167    pragmas(&conn).map_err(|e| e.to_string())?;
168    build(&mut conn, spec.schema, stamp).map_err(|e| e.to_string())?;
169    Ok(conn)
170}
pito_sync/store.rs

pito_sync/
store.rs
