kernoid_sync/
state.rs

1// SPDX-License-Identifier: MIT
2//! The local state of a folder kept in step, in SQLite: the store's tree as last read with the journal state it
3//! belongs to, per path what both sides were when they last agreed (`plan::Base`), and transfers that broke off. One
4//! database per folder; it lives outside the folder, where the app keeps its data. Names in vaults are kept here
5//! decrypted, as the folder holds them anyway.
6
7use std::{collections::BTreeMap, path::Path};
8
9use std::sync::{Mutex, MutexGuard};
10
11use rusqlite::{Connection, OptionalExtension, params};
12
13use crate::{
14    Error, Result,
15    plan::{Base, Local, RelPath, Remote},
16};
17
18/// A node of the store as read from the server.
19#[derive(Debug, Clone, PartialEq, Eq)]
20pub struct Node {
21    pub id: String,
22    pub parent: Option<String>,
23    pub name: String,
24    pub directory: bool,
25    pub blob: Option<String>,
26    pub size: u64,
27    pub role: Option<String>,
28    /// The vault the node lies in (its own id on a vault's root), and its name there encrypted.
29    pub vault: Option<String>,
30    pub sealed_name: Option<String>,
31    /// In a vault whose key was not at hand: `name` is only the node's id there.
32    pub locked: bool,
33}
34
35/// An upload in pieces under way: the file it was for (size and modification time), the pieces sent, and when it
36/// began – the server keeps uploads for a day.
37#[derive(Debug, Clone, PartialEq, Eq)]
38pub struct Pieces {
39    pub size: u64,
40    pub modified: i128,
41    pub blobs: Vec<String>,
42    pub started: i64,
43}
44
45#[derive(Debug)]
46pub struct State {
47    // Behind a lock, so that a round – which holds the state across awaits – can run on any thread.
48    db: Mutex<Connection>,
49}
50
51impl From<rusqlite::Error> for Error {
52    fn from(error: rusqlite::Error) -> Self {
53        Error::State(error.to_string())
54    }
55}
56
57impl State {
58    pub fn open(path: &Path) -> Result<State> {
59        let db = Connection::open(path)?;
60        db.execute_batch(
61            "PRAGMA journal_mode = WAL;
62             CREATE TABLE IF NOT EXISTS meta (key TEXT PRIMARY KEY, value TEXT NOT NULL);
63             CREATE TABLE IF NOT EXISTS remote (
64                 id TEXT PRIMARY KEY, parent TEXT, name TEXT NOT NULL, directory INTEGER NOT NULL, blob TEXT,
65                 size INTEGER NOT NULL, role TEXT);
66             CREATE TABLE IF NOT EXISTS base (
67                 path TEXT PRIMARY KEY, id TEXT NOT NULL, directory INTEGER NOT NULL, blob TEXT, size INTEGER NOT NULL,
68                 local_directory INTEGER NOT NULL, local_size INTEGER NOT NULL, local_modified TEXT NOT NULL);",
69        )?;
70        // 1 (0.3.0): vaults, file ids, transfers.
71        let version: i64 = db.query_row("PRAGMA user_version", [], |row| row.get(0))?;
72        if version < 1 {
73            db.execute_batch(
74                "BEGIN;
75                 ALTER TABLE remote ADD COLUMN vault TEXT;
76                 ALTER TABLE remote ADD COLUMN sealed_name TEXT;
77                 ALTER TABLE remote ADD COLUMN locked INTEGER NOT NULL DEFAULT 0;
78                 ALTER TABLE base ADD COLUMN vault TEXT;
79                 ALTER TABLE base ADD COLUMN local_file TEXT NOT NULL DEFAULT '0';
80                 CREATE TABLE downloads (path TEXT PRIMARY KEY, blob TEXT NOT NULL);
81                 CREATE TABLE uploads (
82                     path TEXT PRIMARY KEY, size INTEGER NOT NULL, modified TEXT NOT NULL, blobs TEXT NOT NULL,
83                     started INTEGER NOT NULL);
84                 PRAGMA user_version = 1;
85                 COMMIT;",
86            )?;
87        }
88        Ok(State { db: Mutex::new(db) })
89    }
90
91    fn db(&self) -> MutexGuard<'_, Connection> {
92        // A round that panicked left nothing half-written: SQLite's transactions saw to that.
93        self.db.lock().unwrap_or_else(|poisoned| poisoned.into_inner())
94    }
95
96    pub fn meta(&self, key: &str) -> Result<Option<String>> {
97        Ok(self.db().query_row("SELECT value FROM meta WHERE key = ?1", [key], |row| row.get(0)).optional()?)
98    }
99
100    pub fn set_meta(&self, key: &str, value: &str) -> Result<()> {
101        self.db().execute("INSERT OR REPLACE INTO meta (key, value) VALUES (?1, ?2)", [key, value])?;
102        Ok(())
103    }
104
105    pub fn remote(&self) -> Result<BTreeMap<String, Node>> {
106        let db = self.db();
107        let mut query =
108            db.prepare("SELECT id, parent, name, directory, blob, size, role, vault, sealed_name, locked FROM remote")?;
109        let rows = query.query_map([], |row| {
110            Ok(Node {
111                id: row.get(0)?,
112                parent: row.get(1)?,
113                name: row.get(2)?,
114                directory: row.get(3)?,
115                blob: row.get(4)?,
116                size: row.get::<_, i64>(5)? as u64,
117                role: row.get(6)?,
118                vault: row.get(7)?,
119                sealed_name: row.get(8)?,
120                locked: row.get(9)?,
121            })
122        })?;
123        let mut out = BTreeMap::new();
124        for node in rows {
125            let node = node?;
126            out.insert(node.id.clone(), node);
127        }
128        Ok(out)
129    }
130
131    /// Replaces the stored tree and its state at once.
132    pub fn set_remote(&mut self, nodes: &BTreeMap<String, Node>, state: &str) -> Result<()> {
133        let mut db = self.db();
134        let tx = db.transaction()?;
135        tx.execute("DELETE FROM remote", [])?;
136        {
137            let mut insert = tx.prepare(
138                "INSERT INTO remote (id, parent, name, directory, blob, size, role, vault, sealed_name, locked)
139                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
140            )?;
141            for node in nodes.values() {
142                insert.execute(params![
143                    node.id,
144                    node.parent,
145                    node.name,
146                    node.directory,
147                    node.blob,
148                    node.size as i64,
149                    node.role,
150                    node.vault,
151                    node.sealed_name,
152                    node.locked
153                ])?;
154            }
155        }
156        tx.execute("INSERT OR REPLACE INTO meta (key, value) VALUES ('state', ?1)", [state])?;
157        tx.commit()?;
158        Ok(())
159    }
160
161    pub fn base(&self) -> Result<BTreeMap<RelPath, Base>> {
162        let db = self.db();
163        let mut query = db.prepare(
164            "SELECT path, id, directory, blob, size, local_directory, local_size, local_modified, vault, local_file
165                 FROM base",
166        )?;
167        let rows = query.query_map([], |row| {
168            let modified: String = row.get(7)?;
169            let file: String = row.get(9)?;
170            Ok((
171                row.get::<_, String>(0)?,
172                Base {
173                    remote: Remote {
174                        id: row.get(1)?,
175                        directory: row.get(2)?,
176                        blob: row.get(3)?,
177                        size: row.get::<_, i64>(4)? as u64,
178                        vault: row.get(8)?,
179                    },
180                    local: Local {
181                        directory: row.get(5)?,
182                        size: row.get::<_, i64>(6)? as u64,
183                        modified: modified.parse().unwrap_or_default(),
184                        file: file.parse().unwrap_or_default(),
185                    },
186                },
187            ))
188        })?;
189        let mut out = BTreeMap::new();
190        for row in rows {
191            let (path, base) = row?;
192            out.insert(path, base);
193        }
194        Ok(out)
195    }
196
197    /// Replaces what both sides agreed on.
198    pub fn set_base(&mut self, base: &BTreeMap<RelPath, Base>) -> Result<()> {
199        let mut db = self.db();
200        let tx = db.transaction()?;
201        tx.execute("DELETE FROM base", [])?;
202        {
203            let mut insert = tx.prepare(
204                "INSERT INTO base
205                   (path, id, directory, blob, size, local_directory, local_size, local_modified, vault, local_file)
206                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
207            )?;
208            for (path, base) in base {
209                insert.execute(params![
210                    path,
211                    base.remote.id,
212                    base.remote.directory,
213                    base.remote.blob,
214                    base.remote.size as i64,
215                    base.local.directory,
216                    base.local.size as i64,
217                    base.local.modified.to_string(),
218                    base.remote.vault,
219                    base.local.file.to_string()
220                ])?;
221            }
222        }
223        tx.commit()?;
224        Ok(())
225    }
226
227    /// The blob a download into a path's partial file was for, if one broke off.
228    pub fn download(&self, path: &str) -> Result<Option<String>> {
229        Ok(self.db().query_row("SELECT blob FROM downloads WHERE path = ?1", [path], |row| row.get(0)).optional()?)
230    }
231
232    /// Records a download begun (`Some`) or ended (`None`).
233    pub fn set_download(&self, path: &str, blob: Option<&str>) -> Result<()> {
234        match blob {
235            Some(blob) => {
236                self.db().execute("INSERT OR REPLACE INTO downloads (path, blob) VALUES (?1, ?2)", [path, blob])?
237            }
238            None => self.db().execute("DELETE FROM downloads WHERE path = ?1", [path])?,
239        };
240        Ok(())
241    }
242
243    /// The paths of downloads that broke off.
244    pub fn downloads(&self) -> Result<Vec<RelPath>> {
245        let db = self.db();
246        let mut query = db.prepare("SELECT path FROM downloads")?;
247        let rows = query.query_map([], |row| row.get(0))?;
248        Ok(rows.collect::<std::result::Result<_, _>>()?)
249    }
250
251    /// The pieces of an upload of a path sent so far.
252    pub fn upload(&self, path: &str) -> Result<Option<Pieces>> {
253        let row = self
254            .db()
255            .query_row("SELECT size, modified, blobs, started FROM uploads WHERE path = ?1", [path], |row| {
256                Ok((row.get::<_, i64>(0)?, row.get::<_, String>(1)?, row.get::<_, String>(2)?, row.get::<_, i64>(3)?))
257            })
258            .optional()?;
259        Ok(row.map(|(size, modified, blobs, started)| Pieces {
260            size: size as u64,
261            modified: modified.parse().unwrap_or_default(),
262            blobs: serde_json::from_str(&blobs).unwrap_or_default(),
263            started,
264        }))
265    }
266
267    /// Records the pieces of an upload (`Some`), or that it ended (`None`).
268    pub fn set_upload(&self, path: &str, pieces: Option<&Pieces>) -> Result<()> {
269        match pieces {
270            Some(p) => self.db().execute(
271                "INSERT OR REPLACE INTO uploads (path, size, modified, blobs, started) VALUES (?1, ?2, ?3, ?4, ?5)",
272                params![
273                    path,
274                    p.size as i64,
275                    p.modified.to_string(),
276                    serde_json::to_string(&p.blobs).unwrap_or_default(),
277                    p.started
278                ],
279            )?,
280            None => self.db().execute("DELETE FROM uploads WHERE path = ?1", [path])?,
281        };
282        Ok(())
283    }
284}
285
286#[cfg(test)]
287mod tests {
288    use super::*;
289
290    #[test]
291    fn what_is_stored_comes_back() {
292        let dir = std::env::temp_dir().join(format!("kernoid-sync-state-{}", std::process::id()));
293        std::fs::create_dir_all(&dir).unwrap();
294        let path = dir.join("state.db");
295        let mut state = State::open(&path).unwrap();
296        let node = Node {
297            id: "n1".into(),
298            parent: None,
299            name: "a.txt".into(),
300            directory: false,
301            blob: Some("b1".into()),
302            size: 3,
303            role: None,
304            vault: Some("v".into()),
305            sealed_name: Some("c2VhbGVk".into()),
306            locked: false,
307        };
308        state.set_remote(&BTreeMap::from([("n1".to_owned(), node.clone())]), "7").unwrap();
309        let base = Base {
310            remote: Remote {
311                id: "n1".into(),
312                directory: false,
313                blob: Some("b1".into()),
314                size: 3,
315                vault: Some("v".into()),
316            },
317            local: Local { directory: false, size: 3, modified: 1_761_000_000_123_456_789, file: u64::MAX },
318        };
319        state.set_base(&BTreeMap::from([("a.txt".to_owned(), base.clone())])).unwrap();
320        drop(state);
321        let state = State::open(&path).unwrap();
322        assert_eq!(state.meta("state").unwrap().as_deref(), Some("7"));
323        assert_eq!(state.remote().unwrap()["n1"], node);
324        assert_eq!(state.base().unwrap()["a.txt"], base);
325        // Transfers that broke off.
326        state.set_download("a.txt", Some("b2")).unwrap();
327        assert_eq!(state.download("a.txt").unwrap().as_deref(), Some("b2"));
328        assert_eq!(state.downloads().unwrap(), ["a.txt"]);
329        state.set_download("a.txt", None).unwrap();
330        assert_eq!(state.download("a.txt").unwrap(), None);
331        let pieces = Pieces { size: 20, modified: 5, blobs: vec!["p1".into(), "p2".into()], started: 1 };
332        state.set_upload("big.bin", Some(&pieces)).unwrap();
333        assert_eq!(state.upload("big.bin").unwrap(), Some(pieces));
334        state.set_upload("big.bin", None).unwrap();
335        assert_eq!(state.upload("big.bin").unwrap(), None);
336        std::fs::remove_dir_all(&dir).unwrap();
337    }
338}
kernoid_sync/state.rs

kernoid_sync/
state.rs
